-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add Cassandra scaler and tests (#2211)
Signed-off-by: nilayasiktoprak <nilayasiktoprak@gmail.com>
- Loading branch information
1 parent
49d60dc
commit cfb18c5
Showing
7 changed files
with
581 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
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
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,229 @@ | ||
| package scalers | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strconv" | ||
| "strings" | ||
|
|
||
| "github.com/gocql/gocql" | ||
| v2beta2 "k8s.io/api/autoscaling/v2beta2" | ||
| "k8s.io/apimachinery/pkg/api/resource" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/apimachinery/pkg/labels" | ||
| "k8s.io/metrics/pkg/apis/external_metrics" | ||
| logf "sigs.k8s.io/controller-runtime/pkg/log" | ||
|
|
||
| kedautil "github.com/kedacore/keda/v2/pkg/util" | ||
| ) | ||
|
|
||
| // cassandraScaler exposes a data pointer to CassandraMetadata and gocql.Session connection. | ||
| type cassandraScaler struct { | ||
| metadata *CassandraMetadata | ||
| session *gocql.Session | ||
| } | ||
|
|
||
| // CassandraMetadata defines metadata used by KEDA to query a Cassandra table. | ||
| type CassandraMetadata struct { | ||
| username string | ||
| password string | ||
| clusterIPAddress string | ||
| port int | ||
| consistency gocql.Consistency | ||
| protocolVersion int | ||
| keyspace string | ||
| query string | ||
| targetQueryValue int | ||
| metricName string | ||
| scalerIndex int | ||
| } | ||
|
|
||
| var cassandraLog = logf.Log.WithName("cassandra_scaler") | ||
|
|
||
| // NewCassandraScaler creates a new Cassandra scaler. | ||
| func NewCassandraScaler(config *ScalerConfig) (Scaler, error) { | ||
| meta, err := ParseCassandraMetadata(config) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error parsing cassandra metadata: %s", err) | ||
| } | ||
|
|
||
| session, err := NewCassandraSession(meta) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error establishing cassandra session: %s", err) | ||
| } | ||
|
|
||
| return &cassandraScaler{ | ||
| metadata: meta, | ||
| session: session, | ||
| }, nil | ||
| } | ||
|
|
||
| // ParseCassandraMetadata parses the metadata and returns a CassandraMetadata or an error if the ScalerConfig is invalid. | ||
| func ParseCassandraMetadata(config *ScalerConfig) (*CassandraMetadata, error) { | ||
| meta := CassandraMetadata{} | ||
|
|
||
| if val, ok := config.TriggerMetadata["query"]; ok { | ||
| meta.query = val | ||
| } else { | ||
| return nil, fmt.Errorf("no query given") | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["targetQueryValue"]; ok { | ||
| targetQueryValue, err := strconv.Atoi(val) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("targetQueryValue parsing error %s", err.Error()) | ||
| } | ||
| meta.targetQueryValue = targetQueryValue | ||
| } else { | ||
| return nil, fmt.Errorf("no targetQueryValue given") | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["username"]; ok { | ||
| meta.username = val | ||
| } else { | ||
| return nil, fmt.Errorf("no username given") | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["port"]; ok { | ||
| port, err := strconv.Atoi(val) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("port parsing error %s", err.Error()) | ||
| } | ||
| meta.port = port | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["clusterIPAddress"]; ok { | ||
| switch p := meta.port; { | ||
| case p > 0: | ||
| meta.clusterIPAddress = fmt.Sprintf("%s:%d", val, meta.port) | ||
| case strings.Contains(val, ":"): | ||
| meta.clusterIPAddress = val | ||
| default: | ||
| return nil, fmt.Errorf("no port given") | ||
| } | ||
| } else { | ||
| return nil, fmt.Errorf("no cluster IP address given") | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["protocolVersion"]; ok { | ||
| protocolVersion, err := strconv.Atoi(val) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("protocolVersion parsing error %s", err.Error()) | ||
| } | ||
| meta.protocolVersion = protocolVersion | ||
| } else { | ||
| meta.protocolVersion = 4 | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["consistency"]; ok { | ||
| meta.consistency = gocql.ParseConsistency(val) | ||
| } else { | ||
| meta.consistency = gocql.One | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["keyspace"]; ok { | ||
| meta.keyspace = val | ||
| } else { | ||
| return nil, fmt.Errorf("no keyspace given") | ||
| } | ||
|
|
||
| if val, ok := config.TriggerMetadata["metricName"]; ok { | ||
| meta.metricName = kedautil.NormalizeString(fmt.Sprintf("cassandra-%s", val)) | ||
| } else { | ||
| meta.metricName = kedautil.NormalizeString(fmt.Sprintf("cassandra-%s", meta.keyspace)) | ||
| } | ||
|
|
||
| if val, ok := config.AuthParams["password"]; ok { | ||
| meta.password = val | ||
| } else { | ||
| return nil, fmt.Errorf("no password given") | ||
| } | ||
|
|
||
| meta.scalerIndex = config.ScalerIndex | ||
|
|
||
| return &meta, nil | ||
| } | ||
|
|
||
| // NewCassandraSession returns a new Cassandra session for the provided CassandraMetadata. | ||
| func NewCassandraSession(meta *CassandraMetadata) (*gocql.Session, error) { | ||
| cluster := gocql.NewCluster(meta.clusterIPAddress) | ||
| cluster.ProtoVersion = meta.protocolVersion | ||
| cluster.Consistency = meta.consistency | ||
| cluster.Authenticator = gocql.PasswordAuthenticator{ | ||
| Username: meta.username, | ||
| Password: meta.password, | ||
| } | ||
|
|
||
| session, err := cluster.CreateSession() | ||
| if err != nil { | ||
| cassandraLog.Error(err, "found error creating session") | ||
| return nil, err | ||
| } | ||
|
|
||
| return session, nil | ||
| } | ||
|
|
||
| // IsActive returns true if there are pending events to be processed. | ||
| func (s *cassandraScaler) IsActive(ctx context.Context) (bool, error) { | ||
| messages, err := s.GetQueryResult(ctx) | ||
| if err != nil { | ||
| return false, fmt.Errorf("error inspecting cassandra: %s", err) | ||
| } | ||
|
|
||
| return messages > 0, nil | ||
| } | ||
|
|
||
| // GetMetricSpecForScaling returns the MetricSpec for the Horizontal Pod Autoscaler. | ||
| func (s *cassandraScaler) GetMetricSpecForScaling() []v2beta2.MetricSpec { | ||
| targetQueryValue := resource.NewQuantity(int64(s.metadata.targetQueryValue), resource.DecimalSI) | ||
| externalMetric := &v2beta2.ExternalMetricSource{ | ||
| Metric: v2beta2.MetricIdentifier{ | ||
| Name: GenerateMetricNameWithIndex(s.metadata.scalerIndex, s.metadata.metricName), | ||
| }, | ||
| Target: v2beta2.MetricTarget{ | ||
| Type: v2beta2.AverageValueMetricType, | ||
| AverageValue: targetQueryValue, | ||
| }, | ||
| } | ||
| metricSpec := v2beta2.MetricSpec{ | ||
| External: externalMetric, Type: externalMetricType, | ||
| } | ||
|
|
||
| return []v2beta2.MetricSpec{metricSpec} | ||
| } | ||
|
|
||
| // GetMetrics returns a value for a supported metric or an error if there is a problem getting the metric. | ||
| func (s *cassandraScaler) GetMetrics(ctx context.Context, metricName string, metricSelector labels.Selector) ([]external_metrics.ExternalMetricValue, error) { | ||
| num, err := s.GetQueryResult(ctx) | ||
| if err != nil { | ||
| return []external_metrics.ExternalMetricValue{}, fmt.Errorf("error inspecting cassandra: %s", err) | ||
| } | ||
|
|
||
| metric := external_metrics.ExternalMetricValue{ | ||
| MetricName: metricName, | ||
| Value: *resource.NewQuantity(int64(num), resource.DecimalSI), | ||
| Timestamp: metav1.Now(), | ||
| } | ||
|
|
||
| return append([]external_metrics.ExternalMetricValue{}, metric), nil | ||
| } | ||
|
|
||
| // GetQueryResult returns the result of the scaler query. | ||
| func (s *cassandraScaler) GetQueryResult(ctx context.Context) (int, error) { | ||
| var value int | ||
| if err := s.session.Query(s.metadata.query).WithContext(ctx).Scan(&value); err != nil { | ||
| if err != gocql.ErrNotFound { | ||
| cassandraLog.Error(err, "query failed") | ||
| return 0, err | ||
| } | ||
| } | ||
|
|
||
| return value, nil | ||
| } | ||
|
|
||
| // Close closes the Cassandra session connection. | ||
| func (s *cassandraScaler) Close() error { | ||
| s.session.Close() | ||
|
|
||
| return nil | ||
| } |
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,79 @@ | ||
| package scalers | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/gocql/gocql" | ||
| ) | ||
|
|
||
| type parseCassandraMetadataTestData struct { | ||
| metadata map[string]string | ||
| isError bool | ||
| authParams map[string]string | ||
| } | ||
|
|
||
| type cassandraMetricIdentifier struct { | ||
| metadataTestData *parseCassandraMetadataTestData | ||
| scalerIndex int | ||
| name string | ||
| } | ||
|
|
||
| var testCassandraMetadata = []parseCassandraMetadataTestData{ | ||
| // nothing passed | ||
| {map[string]string{}, true, map[string]string{}}, | ||
| // everything is passed in verbatim | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "username": "cassandra", "port": "9042", "clusterIPAddress": "cassandra.test", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, false, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no metricName passed, metricName is generated from keyspace | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "username": "cassandra", "clusterIPAddress": "cassandra.test:9042", "keyspace": "test_keyspace", "ScalerIndex": "0"}, false, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no query passed | ||
| {map[string]string{"targetQueryValue": "1", "username": "cassandra", "clusterIPAddress": "cassandra.test:9042", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no targetQueryValue passed | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "username": "cassandra", "clusterIPAddress": "cassandra.test:9042", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no username passed | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "clusterIPAddress": "cassandra.test:9042", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no port passed | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "username": "cassandra", "clusterIPAddress": "cassandra.test", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no clusterIPAddress passed | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "username": "cassandra", "port": "9042", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no keyspace passed | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "username": "cassandra", "clusterIPAddress": "cassandra.test:9042", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{"password": "Y2Fzc2FuZHJhCg=="}}, | ||
| // no password passed | ||
| {map[string]string{"query": "SELECT COUNT(*) FROM test_keyspace.test_table;", "targetQueryValue": "1", "username": "cassandra", "clusterIPAddress": "cassandra.test:9042", "keyspace": "test_keyspace", "ScalerIndex": "0", "metricName": "myMetric"}, true, map[string]string{}}, | ||
| } | ||
|
|
||
| var cassandraMetricIdentifiers = []cassandraMetricIdentifier{ | ||
| {&testCassandraMetadata[1], 0, "s0-cassandra-myMetric"}, | ||
| {&testCassandraMetadata[2], 1, "s1-cassandra-test_keyspace"}, | ||
| } | ||
|
|
||
| func TestCassandraParseMetadata(t *testing.T) { | ||
| testCaseNum := 1 | ||
| for _, testData := range testCassandraMetadata { | ||
| _, err := ParseCassandraMetadata(&ScalerConfig{TriggerMetadata: testData.metadata, AuthParams: testData.authParams}) | ||
| if err != nil && !testData.isError { | ||
| t.Errorf("Expected success but got error for unit test # %v", testCaseNum) | ||
| } | ||
| if testData.isError && err == nil { | ||
| t.Errorf("Expected error but got success for unit test # %v", testCaseNum) | ||
| } | ||
| testCaseNum++ | ||
| } | ||
| } | ||
|
|
||
| func TestCassandraGetMetricSpecForScaling(t *testing.T) { | ||
| for _, testData := range cassandraMetricIdentifiers { | ||
| meta, err := ParseCassandraMetadata(&ScalerConfig{TriggerMetadata: testData.metadataTestData.metadata, ScalerIndex: testData.scalerIndex, AuthParams: testData.metadataTestData.authParams}) | ||
| if err != nil { | ||
| t.Fatal("Could not parse metadata:", err) | ||
| } | ||
| cluster := gocql.NewCluster(meta.clusterIPAddress) | ||
| session, _ := cluster.CreateSession() | ||
| mockCassandraScaler := cassandraScaler{meta, session} | ||
|
|
||
| metricSpec := mockCassandraScaler.GetMetricSpecForScaling() | ||
| metricName := metricSpec[0].External.Metric.Name | ||
| if metricName != testData.name { | ||
| t.Errorf("Wrong External metric source name: %s, expected: %s", metricName, testData.name) | ||
| } | ||
| } | ||
| } |
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.