Files
tasks/common/persistence/tests/cassandra_test.go
pdoernerandGitHub 68a1d21755 Dynamic rate limiter (#4390)
* moving window average

* remove channel avg impl

* add signal aggregator

* adjust record fn

* add health signal clients

* inject signal aggregator

* fix tests

* add metric emission

* add health request rate limiter

* WIP

* race condition

* WIP

* add health request rate limiter

* bench test rate limiter

* WIP

* Revert "race condition"

This reverts commit e283bfb3dca7228d715cd28235e436caafb34fa6.

* Revert "add metric emission"

This reverts commit 859950ee2965b5ac2144f46930e1308d95890e69.

* emit per shard RPS metric

* cleanup

* merge metric and signal clients

* cleanup

* cleanup

* linting

* global dynamic rate limiter

* cleanup

* remove generics

* WIP

* cleanup

* fix deferred metric fn

* fix defer metric fn

* fix clients

* types

* simple dynamic rate limiter test

* linting

* tests

* tests

* acquire lock once

* locks

* array moving average

* Revert "array moving average"

This reverts commit 98f2b66b9b7a711c427d5f55de82bc55d135ddfc.

* cleanup

* emit per shard RPS

* array average

* feedback

* cleanup

* handle nil health signals

* tests

* uncomment signal collection

* refactor dynamic config properties

* cleanup

* add aggregation feature flag

* fix test

* cleanup

* avoid potential race condition

* feedback

* fix

* remove test
2023-06-02 15:30:01 -07:00

175 lines
5.0 KiB
Go

// The MIT License
//
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
//
// Copyright (c) 2020 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package tests
import (
"testing"
"github.com/stretchr/testify/suite"
persistencetests "go.temporal.io/server/common/persistence/persistence-tests"
"go.temporal.io/server/common/persistence/serialization"
_ "go.temporal.io/server/common/persistence/sql/sqlplugin/mysql"
)
func TestCassandraShardStoreSuite(t *testing.T) {
testData, tearDown := setUpCassandraTest(t)
defer tearDown()
shardStore, err := testData.Factory.NewShardStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
s := NewShardSuite(
t,
shardStore,
serialization.NewSerializer(),
testData.Logger,
)
suite.Run(t, s)
}
func TestCassandraExecutionMutableStateStoreSuite(t *testing.T) {
testData, tearDown := setUpCassandraTest(t)
defer tearDown()
shardStore, err := testData.Factory.NewShardStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
executionStore, err := testData.Factory.NewExecutionStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
s := NewExecutionMutableStateSuite(
t,
shardStore,
executionStore,
serialization.NewSerializer(),
testData.Logger,
)
suite.Run(t, s)
}
func TestCassandraExecutionMutableStateTaskStoreSuite(t *testing.T) {
testData, tearDown := setUpCassandraTest(t)
defer tearDown()
shardStore, err := testData.Factory.NewShardStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
executionStore, err := testData.Factory.NewExecutionStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
s := NewExecutionMutableStateTaskSuite(
t,
shardStore,
executionStore,
serialization.NewSerializer(),
testData.Logger,
)
suite.Run(t, s)
}
func TestCassandraHistoryStoreSuite(t *testing.T) {
testData, tearDown := setUpCassandraTest(t)
defer tearDown()
store, err := testData.Factory.NewExecutionStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
s := NewHistoryEventsSuite(t, store, testData.Logger)
suite.Run(t, s)
}
func TestCassandraTaskQueueSuite(t *testing.T) {
testData, tearDown := setUpCassandraTest(t)
defer tearDown()
taskQueueStore, err := testData.Factory.NewTaskStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
s := NewTaskQueueSuite(t, taskQueueStore, testData.Logger)
suite.Run(t, s)
}
func TestCassandraTaskQueueTaskSuite(t *testing.T) {
testData, tearDown := setUpCassandraTest(t)
defer tearDown()
taskQueueStore, err := testData.Factory.NewTaskStore()
if err != nil {
t.Fatalf("unable to create Cassandra DB: %v", err)
}
s := NewTaskQueueTaskSuite(t, taskQueueStore, testData.Logger)
suite.Run(t, s)
}
func TestCassandraVisibilityPersistence(t *testing.T) {
s := &VisibilityPersistenceSuite{
TestBase: persistencetests.NewTestBaseWithCassandra(&persistencetests.TestBaseOptions{}),
}
suite.Run(t, s)
}
// TODO: Merge persistence-tests into the tests directory.
func TestCassandraHistoryV2Persistence(t *testing.T) {
s := new(persistencetests.HistoryV2PersistenceSuite)
s.TestBase = persistencetests.NewTestBaseWithCassandra(&persistencetests.TestBaseOptions{})
s.TestBase.Setup(nil)
suite.Run(t, s)
}
func TestCassandraMetadataPersistenceV2(t *testing.T) {
s := new(persistencetests.MetadataPersistenceSuiteV2)
s.TestBase = persistencetests.NewTestBaseWithCassandra(&persistencetests.TestBaseOptions{})
s.TestBase.Setup(nil)
suite.Run(t, s)
}
func TestCassandraQueuePersistence(t *testing.T) {
s := new(persistencetests.QueuePersistenceSuite)
s.TestBase = persistencetests.NewTestBaseWithCassandra(&persistencetests.TestBaseOptions{})
s.TestBase.Setup(nil)
suite.Run(t, s)
}
func TestCassandraClusterMetadataPersistence(t *testing.T) {
s := new(persistencetests.ClusterMetadataManagerSuite)
s.TestBase = persistencetests.NewTestBaseWithCassandra(&persistencetests.TestBaseOptions{})
s.TestBase.Setup(nil)
suite.Run(t, s)
}