Changed cachine implementation. Fixes issue #5
- Creates tenant cache from snapshot instead of DB
- updates tenant cache on changeList event or if a scope if not in cache
then update info from DB
- Does not create developer Info cache during init
- Instead on a miss, get info from db and populate cache
- This allows us to maintain cache only for active apps
- Invalidate this cache every configured interval
- Set useCaching default to false
- Set default buffer size to 1000
diff --git a/README.md b/README.md
index 0b41867..699a2f6 100644
--- a/README.md
+++ b/README.md
@@ -14,20 +14,21 @@
| apidanalytics_uap_server_base | string. url. required. |
| apidanalytics_use_caching | boolean. default: true |
| apidanalytics_buffer_channel_size | int. number of slots. default: 100|
+| apidanalytics_cache_refresh_interval | int. seconds. default: 1800 |
### Startup Procedure
1. Initialize crash recovery, upload and buffering manager to handle buffering analytics messages to files
locally and then periodically upload these files to S3/GCS based on signedURL received from
uapCollectionEndpoint exposed via edgex proxy
2. Create a listener for Apigee-Sync event
- 1. Each time a Snapshot is received, create an in-memory cache for data scope and developer information
- 2. Each time a changeList is received
- 1. if data_scope info changed, then insert/delete info for changed scope from tenantCache
- 2. if any other kms table is changed, then refresh entire developerInfo cache as only specific
- fields are saved in the cache
+ 1. Each time a Snapshot is received, create an in-memory cache for data scope
+ 2. Each time a changeList is received, if data_scope info changed, then insert/delete info for changed scope from tenantCache
+ 3. Developer info cache will be invalidated periodically and populated when 1st request for that apiKey comes in
3. Initialize POST /analytics/{scope_uuid} API
4. Upon receiving POST requests
1. Validate and enrich each batch of analytics records
+ 1. If developerCache does not have info for apiKey then get from DB and insert into cache.
+ This way the cache will only have info for developers/app with active traffic
2. If valid, then publish records to an internal buffer channel
5. Buffering Logic
1. Buffering manager creates listener on the internal buffer channel and thus consumes messages
diff --git a/apidAnalytics_suite_test.go b/apidAnalytics_suite_test.go
index 3a7e0c4..46e2421 100644
--- a/apidAnalytics_suite_test.go
+++ b/apidAnalytics_suite_test.go
@@ -47,15 +47,6 @@
config.Set(uapServerBase, "http://localhost:9000") // dummy value
Expect(apid.InitializePlugins).ToNot(Panic())
- // create initial cache for tenant and developer info
- config.Set(useCaching, true)
-
- createTenantCache()
- Expect(len(tenantCache)).To(Equal(1))
-
- createDeveloperInfoCache()
- Expect(len(developerInfoCache)).To(Equal(1))
-
// Analytics POST API
router := apid.API().Router()
router.HandleFunc(analyticsBasePath+"/{bundle_scope_uuid}", func(w http.ResponseWriter, req *http.Request) {
diff --git a/common_helper.go b/common_helper.go
index 4b07b66..91a85a2 100644
--- a/common_helper.go
+++ b/common_helper.go
@@ -2,7 +2,7 @@
import (
"database/sql"
- "fmt"
+ "github.com/apigee-labs/transicator/common"
"sync"
)
@@ -22,78 +22,40 @@
// Load data scope information into an in-memory cache so that
// for each record a DB lookup is not required
-func createTenantCache() error {
+func createTenantCache(snapshot *common.Snapshot) {
// Lock before writing to the map as it has multiple readers
tenantCachelock.Lock()
defer tenantCachelock.Unlock()
tenantCache = make(map[string]tenant)
- var org, env, tenantId, id string
- db := getDB()
- rows, error := db.Query("SELECT env, org, scope, id FROM DATA_SCOPE")
+ for _, table := range snapshot.Tables {
+ switch table.Name {
+ case "edgex.data_scope":
+ for _, row := range table.Rows {
+ var org, env, tenantId, id string
- if error != nil {
- return fmt.Errorf("Count not get datascope from "+
- "DB due to: %v", error)
- } else {
- defer rows.Close()
- for rows.Next() {
- rows.Scan(&env, &org, &tenantId, &id)
- tenantCache[id] = tenant{Org: org,
- Env: env,
- TenantId: tenantId}
+ row.Get("id", &id)
+ row.Get("scope", &tenantId)
+ row.Get("org", &org)
+ row.Get("env", &env)
+
+ tenantCache[id] = tenant{Org: org,
+ Env: env,
+ TenantId: tenantId}
+ }
}
}
-
log.Debugf("Count of data scopes in the cache: %d", len(tenantCache))
- return nil
}
// Load data scope information into an in-memory cache so that
// for each record a DB lookup is not required
-func createDeveloperInfoCache() error {
+func updateDeveloperInfoCache() {
// Lock before writing to the map as it has multiple readers
developerInfoCacheLock.Lock()
defer developerInfoCacheLock.Unlock()
developerInfoCache = make(map[string]developerInfo)
- var apiProduct, developerApp, developerEmail, developer sql.NullString
- var tenantId, apiKey string
-
- db := getDB()
- sSql := "SELECT mp.tenant_id, mp.appcred_id, ap.name," +
- " a.name, d.username, d.email " +
- "FROM APP_CREDENTIAL_APIPRODUCT_MAPPER as mp " +
- "INNER JOIN API_PRODUCT as ap ON ap.id = mp.apiprdt_id " +
- "INNER JOIN APP AS a ON a.id = mp.app_id " +
- "INNER JOIN DEVELOPER as d ON d.id = a.developer_id;"
- rows, error := db.Query(sSql)
-
- if error != nil {
- return fmt.Errorf("Count not get developerInfo "+
- "from DB due to: %v", error)
- } else {
- defer rows.Close()
- for rows.Next() {
- rows.Scan(&tenantId, &apiKey, &apiProduct,
- &developerApp, &developer, &developerEmail)
-
- keyForMap := getKeyForDeveloperInfoCache(tenantId, apiKey)
- apiPrd := getValuesIgnoringNull(apiProduct)
- devApp := getValuesIgnoringNull(developerApp)
- dev := getValuesIgnoringNull(developer)
- devEmail := getValuesIgnoringNull(developerEmail)
-
- developerInfoCache[keyForMap] = developerInfo{
- ApiProduct: apiPrd,
- DeveloperApp: devApp,
- DeveloperEmail: devEmail,
- Developer: dev}
- }
- }
-
- log.Debugf("Count of apiKey~tenantId combinations "+
- "in the cache: %d", len(developerInfoCache))
- return nil
+ log.Debug("Invalidated developerInfo cache")
}
// Returns Tenant Info given a scope uuid from the cache or by querying
@@ -102,104 +64,137 @@
if config.GetBool(useCaching) {
// acquire a read lock as this cache has 1 writer as well
tenantCachelock.RLock()
- defer tenantCachelock.RUnlock()
ten, exists := tenantCache[scopeuuid]
+ tenantCachelock.RUnlock()
+ dbErr := dbError{}
+
if !exists {
- reason := "No tenant found for this scopeuuid: " + scopeuuid
- errorCode := "UNKNOWN_SCOPE"
- // Incase of unknown scope, try to refresh the
- // cache ansynchronously incase an update was missed or delayed
- go createTenantCache()
- return tenant{}, dbError{
- ErrorCode: errorCode,
- Reason: reason}
- } else {
- return ten, dbError{}
+ log.Debugf("No tenant found for scopeuuid = %s "+
+ "in cache", scopeuuid)
+ log.Debug("loading info from DB")
+
+ // Update cache
+ t, err := getTenantFromDB(scopeuuid)
+
+ if err.ErrorCode != "" {
+ dbErr = err
+ ten = t
+ } else {
+ // update cache
+ tenantCachelock.Lock()
+ defer tenantCachelock.Unlock()
+ tenantCache[scopeuuid] = t
+ ten = t
+ }
}
+ return ten, dbErr
} else {
- var org, env, tenantId string
-
- db := getDB()
- error := db.QueryRow("SELECT env, org, scope FROM DATA_SCOPE"+
- " where id = ?", scopeuuid).Scan(&env, &org, &tenantId)
-
- switch {
- case error == sql.ErrNoRows:
- reason := "No tenant found for this scopeuuid: " + scopeuuid
- errorCode := "UNKNOWN_SCOPE"
- return tenant{}, dbError{
- ErrorCode: errorCode,
- Reason: reason}
- case error != nil:
- reason := error.Error()
- errorCode := "INTERNAL_SEARCH_ERROR"
- return tenant{}, dbError{
- ErrorCode: errorCode,
- Reason: reason}
- }
- return tenant{
- Org: org,
- Env: env,
- TenantId: tenantId}, dbError{}
+ return getTenantFromDB(scopeuuid)
}
}
-// Returns Dveloper related info given an apiKey and tenantId
+// Returns Developer related info given an apiKey and tenantId
// from the cache or by querying the DB directly based on useCachig config
func getDeveloperInfo(tenantId string, apiKey string) developerInfo {
if config.GetBool(useCaching) {
keyForMap := getKeyForDeveloperInfoCache(tenantId, apiKey)
// acquire a read lock as this cache has 1 writer as well
- tenantCachelock.RLock()
- defer tenantCachelock.RUnlock()
+ developerInfoCacheLock.RLock()
devInfo, exists := developerInfoCache[keyForMap]
+ developerInfoCacheLock.RUnlock()
+
if !exists {
- log.Warnf("No data found for for tenantId = %s"+
- " and apiKey = %s", tenantId, apiKey)
- // Incase of unknown apiKey~tenantId,
- // try to refresh the cache ansynchronously incase an update was missed or delayed
- go createDeveloperInfoCache()
- return developerInfo{}
- } else {
- return devInfo
+ log.Debugf("No data found for for tenantId = %s"+
+ " and apiKey = %s in cache", tenantId, apiKey)
+ log.Debug("loading info from DB")
+
+ // Update cache
+ dev, err := getDevInfoFromDB(tenantId, apiKey)
+
+ if err == nil {
+ // update cache
+ developerInfoCacheLock.Lock()
+ defer developerInfoCacheLock.Unlock()
+ key := getKeyForDeveloperInfoCache(tenantId, apiKey)
+ developerInfoCache[key] = dev
+ }
+
+ devInfo = dev
+
}
+ return devInfo
} else {
- var apiProduct, developerApp, developerEmail sql.NullString
- var developer sql.NullString
-
- db := getDB()
- sSql := "SELECT ap.name, a.name, d.username, d.email " +
- "FROM APP_CREDENTIAL_APIPRODUCT_MAPPER as mp " +
- "INNER JOIN API_PRODUCT as ap ON ap.id = mp.apiprdt_id " +
- "INNER JOIN APP AS a ON a.id = mp.app_id " +
- "INNER JOIN DEVELOPER as d ON d.id = a.developer_id " +
- "where mp.tenant_id = ? and mp.appcred_id = ?;"
- error := db.QueryRow(sSql, tenantId, apiKey).
- Scan(&apiProduct, &developerApp,
- &developer, &developerEmail)
-
- switch {
- case error == sql.ErrNoRows:
- log.Debugf("No data found for for tenantId = %s "+
- "and apiKey = %s", tenantId, apiKey)
- return developerInfo{}
- case error != nil:
- log.Debugf("No data found for for tenantId = %s and "+
- "apiKey = %s due to: %v", tenantId, apiKey, error)
- return developerInfo{}
- }
-
- apiPrd := getValuesIgnoringNull(apiProduct)
- devApp := getValuesIgnoringNull(developerApp)
- dev := getValuesIgnoringNull(developer)
- devEmail := getValuesIgnoringNull(developerEmail)
- return developerInfo{ApiProduct: apiPrd,
- DeveloperApp: devApp,
- DeveloperEmail: devEmail,
- Developer: dev}
+ devInfo, _ := getDevInfoFromDB(tenantId, apiKey)
+ return devInfo
}
}
+// Returns tenant info by querying DB directly
+func getTenantFromDB(scopeuuid string) (tenant, dbError) {
+ var org, env, tenantId string
+
+ db := getDB()
+ error := db.QueryRow("SELECT env, org, scope FROM DATA_SCOPE"+
+ " where id = ?", scopeuuid).Scan(&env, &org, &tenantId)
+
+ switch {
+ case error == sql.ErrNoRows:
+ reason := "No tenant found for this scopeuuid: " + scopeuuid
+ errorCode := "UNKNOWN_SCOPE"
+ return tenant{}, dbError{
+ ErrorCode: errorCode,
+ Reason: reason}
+ case error != nil:
+ reason := error.Error()
+ errorCode := "INTERNAL_SEARCH_ERROR"
+ return tenant{}, dbError{
+ ErrorCode: errorCode,
+ Reason: reason}
+ }
+ return tenant{
+ Org: org,
+ Env: env,
+ TenantId: tenantId}, dbError{}
+}
+
+// Returns developer info by querying DB directly
+func getDevInfoFromDB(tenantId string, apiKey string) (developerInfo, error) {
+ var apiProduct, developerApp, developerEmail sql.NullString
+ var developer sql.NullString
+
+ db := getDB()
+ sSql := "SELECT ap.name, a.name, d.username, d.email " +
+ "FROM APP_CREDENTIAL_APIPRODUCT_MAPPER as mp " +
+ "INNER JOIN API_PRODUCT as ap ON ap.id = mp.apiprdt_id " +
+ "INNER JOIN APP AS a ON a.id = mp.app_id " +
+ "INNER JOIN DEVELOPER as d ON d.id = a.developer_id " +
+ "where mp.tenant_id = ? and mp.appcred_id = ?;"
+ error := db.QueryRow(sSql, tenantId, apiKey).
+ Scan(&apiProduct, &developerApp,
+ &developer, &developerEmail)
+
+ switch {
+ case error == sql.ErrNoRows:
+ log.Debugf("No data found for for tenantId = %s "+
+ "and apiKey = %s in DB", tenantId, apiKey)
+ return developerInfo{}, error
+ case error != nil:
+ log.Debugf("No data found for for tenantId = %s and "+
+ "apiKey = %s due to: %v", tenantId, apiKey, error)
+ return developerInfo{}, error
+ }
+
+ apiPrd := getValuesIgnoringNull(apiProduct)
+ devApp := getValuesIgnoringNull(developerApp)
+ dev := getValuesIgnoringNull(developer)
+ devEmail := getValuesIgnoringNull(developerEmail)
+
+ return developerInfo{ApiProduct: apiPrd,
+ DeveloperApp: devApp,
+ DeveloperEmail: devEmail,
+ Developer: dev}, nil
+}
+
// Helper method to handle scanning null values in DB to empty string
func getValuesIgnoringNull(sqlValue sql.NullString) string {
if sqlValue.Valid {
diff --git a/common_helper_test.go b/common_helper_test.go
index 31bdb31..5e23618 100644
--- a/common_helper_test.go
+++ b/common_helper_test.go
@@ -4,10 +4,21 @@
"database/sql"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
+
+ "github.com/apigee-labs/transicator/common"
)
var _ = Describe("test getTenantForScope()", func() {
Context("with usecaching set to true", func() {
+ BeforeEach(func() {
+ config.Set(useCaching, true)
+ snapshot := getDatascopeSnapshot()
+ createTenantCache(&snapshot)
+ Expect(len(tenantCache)).To(Equal(1))
+ })
+ AfterEach(func() {
+ config.Set(useCaching, false)
+ })
Context("get tenant for valid scopeuuid", func() {
It("should return testorg and testenv", func() {
tenant, dbError := getTenantForScope("testid")
@@ -26,12 +37,6 @@
})
Context("with usecaching set to false", func() {
- BeforeEach(func() {
- config.Set(useCaching, false)
- })
- AfterEach(func() {
- config.Set(useCaching, true)
- })
Context("get tenant for valid scopeuuid", func() {
It("should return testorg and testenv", func() {
tenant, dbError := getTenantForScope("testid")
@@ -50,18 +55,49 @@
})
})
+var _ = Describe("test getTenantFromDB()", func() {
+ Context("get developerInfo for valid scopeuuid", func() {
+ It("should return all right data", func() {
+ tenant, dbError := getTenantFromDB("testid")
+ Expect(dbError.Reason).To(Equal(""))
+ Expect(tenant.TenantId).To(Equal("tenantid"))
+ })
+ })
+ Context("get developerInfo for invalid scopeuuid", func() {
+ It("should return error", func() {
+ tenant, dbError := getTenantFromDB("wrongid")
+ Expect(tenant.Org).To(Equal(""))
+ Expect(dbError.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
+ })
+ })
+
+})
+
var _ = Describe("test getDeveloperInfo()", func() {
Context("with usecaching set to true", func() {
+ BeforeEach(func() {
+ config.Set(useCaching, true)
+ updateDeveloperInfoCache()
+ })
+ AfterEach(func() {
+ config.Set(useCaching, false)
+ })
Context("get developerInfo for valid tenantId and apikey", func() {
It("should return all right data", func() {
- developerInfo := getDeveloperInfo("tenantid", "testapikey")
- Expect(developerInfo.ApiProduct).To(Equal("testproduct"))
- Expect(developerInfo.Developer).To(Equal("testdeveloper"))
+ key := getKeyForDeveloperInfoCache("tenantid", "testapikey")
+ _, e := developerInfoCache[key]
+ Expect(e).To(BeFalse())
+
+ getDeveloperInfo("tenantid", "testapikey")
+ devInfo, e := developerInfoCache[key]
+ Expect(e).To(BeTrue())
+ Expect(devInfo.ApiProduct).To(Equal("testproduct"))
+ Expect(devInfo.Developer).To(Equal("testdeveloper"))
})
})
Context("get developerInfo for invalid tenantId and apikey", func() {
- It("should return all right data", func() {
+ It("should return all empty", func() {
developerInfo := getDeveloperInfo("wrongid", "wrongapikey")
Expect(developerInfo.ApiProduct).To(Equal(""))
})
@@ -69,12 +105,6 @@
})
Context("with usecaching set to false", func() {
- BeforeEach(func() {
- config.Set(useCaching, false)
- })
- AfterEach(func() {
- config.Set(useCaching, true)
- })
Context("get developerInfo for valid tenantId and apikey", func() {
It("should return all right data", func() {
developerInfo := getDeveloperInfo("tenantid", "testapikey")
@@ -91,6 +121,25 @@
})
})
+var _ = Describe("test getDevInfoFromDB()", func() {
+ Context("get developerInfo for valid tenantId and apikey", func() {
+ It("should return all right data", func() {
+ developerInfo, err := getDevInfoFromDB("tenantid", "testapikey")
+ Expect(err).ToNot(HaveOccurred())
+ Expect(developerInfo.ApiProduct).To(Equal("testproduct"))
+ Expect(developerInfo.Developer).To(Equal("testdeveloper"))
+ })
+ })
+ Context("get developerInfo for invalid tenantId and apikey", func() {
+ It("should return all empty data", func() {
+ developerInfo, err := getDevInfoFromDB("wrongid", "wrongapikey")
+ Expect(err).To(HaveOccurred())
+ Expect(developerInfo.ApiProduct).To(Equal(""))
+ })
+ })
+
+})
+
var _ = Describe("test getValuesIgnoringNull()", func() {
Context("Null sql value", func() {
It("should return empty string", func() {
@@ -107,3 +156,23 @@
})
})
})
+
+func getDatascopeSnapshot() common.Snapshot {
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot_valid",
+ Tables: []common.Table{
+ {
+ Name: LISTENER_TABLE_DATA_SCOPE,
+ Rows: []common.Row{
+ {
+ "id": &common.ColumnVal{Value: "testid"},
+ "scope": &common.ColumnVal{Value: "tenantid"},
+ "org": &common.ColumnVal{Value: "testorg"},
+ "env": &common.ColumnVal{Value: "testenv"},
+ },
+ },
+ },
+ },
+ }
+ return event
+}
diff --git a/init.go b/init.go
index 451a1f4..0706e57 100644
--- a/init.go
+++ b/init.go
@@ -6,6 +6,7 @@
"os"
"path/filepath"
"sync"
+ "time"
)
const (
@@ -29,7 +30,7 @@
// Number of slots for internal channel buffering of
// analytics records before they are dumped to a file
analyticsBufferChannelSize = "apidanalytics_buffer_channel_size"
- analyticsBufferChannelSizeDefault = 100
+ analyticsBufferChannelSizeDefault = 1000
// EdgeX endpoint base path to access Uap Collection Endpoint
uapServerBase = "apidanalytics_uap_server_base"
@@ -38,7 +39,11 @@
// info will be maintained in-memory
// cache to avoid DB calls for each analytics message
useCaching = "apidanalytics_use_caching"
- useCachingDefault = true
+ useCachingDefault = false
+
+ // Interval in seconds when the developer cache should be refreshed
+ analyticsCacheRefreshInterval = "apidanalytics_cache_refresh_interval"
+ analyticsCacheRefreshIntervaleDefault = 1800
)
// keep track of the services that this plugin will use
@@ -122,6 +127,22 @@
// for new messages and dump them to files
initBufferingManager()
+ // Initialize developerInfo cache invalidation periodically
+ if config.GetBool(useCaching) {
+ updateDeveloperInfoCache()
+ go func() {
+ ticker := time.NewTicker(time.Second *
+ config.GetDuration(analyticsCacheRefreshInterval))
+ // Ticker will keep running till go routine is running
+ // i.e. till application is running
+ defer ticker.Stop()
+
+ for range ticker.C {
+ updateDeveloperInfoCache()
+ }
+ }()
+ }
+
// Initialize API's and expose them
initAPI(services)
log.Debug("end init for apidAnalytics plugin")
@@ -153,6 +174,9 @@
// set default config for useCaching
config.SetDefault(useCaching, useCachingDefault)
+ // set default config for cache refresh interval
+ config.SetDefault(analyticsCacheRefreshInterval, analyticsCacheRefreshIntervaleDefault)
+
// set default config for upload interval
config.SetDefault(analyticsUploadInterval, analyticsUploadIntervalDefault)
diff --git a/listener.go b/listener.go
index ba5e15a..a088687 100644
--- a/listener.go
+++ b/listener.go
@@ -37,21 +37,11 @@
setDB(db)
if config.GetBool(useCaching) {
- err = createTenantCache()
- if err != nil {
- log.Error(err)
- } else {
- log.Debug("Created a local cache" +
- " for datasope information")
- }
- err = createDeveloperInfoCache()
- if err != nil {
- log.Error(err)
- } else {
- log.Debug("Created a local cache for developer information")
- }
+ createTenantCache(snapshot)
+ log.Debug("Created a local cache" +
+ " for datasope information")
} else {
- log.Info("Will not be caching any developer info " +
+ log.Info("Will not be caching any developer or tenant info " +
"and make a DB call for every analytics msg")
}
return
@@ -62,7 +52,6 @@
log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
var rows []common.Row
- refreshDevInfoNeeded := false
for _, payload := range changes.Changes {
rows = nil
switch payload.Table {
@@ -104,17 +93,8 @@
" scope: %s", scopeuuid)
}
}
- case "kms.developer", "kms.app", "kms.api_product",
- "kms.app_credential_apiproduct_mapper":
- // any change in any of the above tables
- // should result in cache refresh
- refreshDevInfoNeeded = true
}
}
- // Refresh cache once for all set of changes
- if refreshDevInfoNeeded {
- createDeveloperInfoCache()
- log.Debug("Refresh local developerInfoCache")
- }
+
}
}
diff --git a/listener_test.go b/listener_test.go
index b792133..279b4f7 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -4,19 +4,86 @@
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
+ "github.com/30x/apid"
"github.com/apigee-labs/transicator/common"
)
const (
- LISTENER_TABLE_APP_CRED_MAPPER = "kms.app_credential_apiproduct_mapper"
- LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope"
+ LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope"
)
-var _ = Describe("listener", func() {
+var _ = Describe("ApigeeSync event", func() {
+
+ var db apid.DB
+ handler := handler{}
+
+ BeforeEach(func() {
+ db = getDB()
+
+ config.Set(useCaching, true)
+
+ snapshot := getDatascopeSnapshot()
+ createTenantCache(&snapshot)
+ Expect(len(tenantCache)).To(Equal(1))
+ })
+
+ AfterEach(func() {
+ config.Set(useCaching, false)
+ setDB(db)
+ })
+
+ Context("ApigeeSync snapshot event", func() {
+ It("should set DB to appropriate version", func() {
+ config.Set(useCaching, false)
+
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot",
+ Tables: []common.Table{},
+ }
+
+ handler.Handle(&event)
+
+ expectedDB, err := data.DBVersion(event.SnapshotInfo)
+ Expect(err).NotTo(HaveOccurred())
+
+ Expect(getDB() == expectedDB).Should(BeTrue())
+ })
+
+ It("should process a valid Snapshot", func() {
+ event := common.Snapshot{
+ SnapshotInfo: "test_snapshot_valid",
+ Tables: []common.Table{
+ {
+ Name: LISTENER_TABLE_DATA_SCOPE,
+ Rows: []common.Row{
+ {
+ "id": &common.ColumnVal{Value: "i"},
+ "_change_selector": &common.ColumnVal{Value: "c"},
+ "apid_cluster_id": &common.ColumnVal{Value: "a"},
+ "scope": &common.ColumnVal{Value: "s"},
+ "org": &common.ColumnVal{Value: "o"},
+ "env": &common.ColumnVal{Value: "e"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ },
+ },
+ },
+ },
+ }
+
+ handler.Handle(&event)
+ tenant := tenantCache["i"]
+ Expect(tenant.TenantId).To(Equal("s"))
+ Expect(tenant.Org).To(Equal("o"))
+ Expect(tenant.Env).To(Equal("e"))
+ })
+ })
+
Context("Process changeList", func() {
Context(LISTENER_TABLE_DATA_SCOPE, func() {
It("insert/delete event should add/remove to/from cache if usecaching is true", func() {
- config.Set(useCaching, true)
txn, err := getDB().Begin()
Expect(err).ShouldNot(HaveOccurred())
txn.Exec("INSERT INTO DATA_SCOPE (id, _change_selector, apid_cluster_id, scope, org, env) "+
@@ -53,7 +120,7 @@
},
}
- processChange(&insert)
+ handler.Handle(&insert)
tenant := tenantCache["i2"]
Expect(tenant.TenantId).To(Equal("s2"))
Expect(tenant.Org).To(Equal("o2"))
@@ -75,65 +142,9 @@
},
}
- processChange(&delete)
+ handler.Handle(&delete)
_, exists := tenantCache["i2"]
Expect(exists).To(Equal(false))
-
- })
- })
- Context(LISTENER_TABLE_APP_CRED_MAPPER, func() {
- It("insert/delete event should refresh developer cache if usecaching is true", func() {
- config.Set(useCaching, true)
-
- txn, err := getDB().Begin()
- Expect(err).ShouldNot(HaveOccurred())
- txn.Exec("INSERT INTO APP_CREDENTIAL_APIPRODUCT_MAPPER (tenant_id, appcred_id, app_id, apiprdt_id, status, _change_selector) "+
- "VALUES"+
- "($1,$2,$3,$4,$5,$6)",
- "tenantid",
- "aci",
- "ai",
- "testproductid",
- "APPROVED",
- "12345",
- )
-
- txn.Exec("INSERT INTO APP (id, tenant_id, name, developer_id) "+
- "VALUES"+
- "($1,$2,$3,$4)",
- "ai",
- "tenantid",
- "name",
- "testdeveloperid",
- )
- txn.Commit()
-
- insert := common.ChangeList{
- LastSequence: "test",
- Changes: []common.Change{
- {
- Operation: common.Insert,
- Table: LISTENER_TABLE_APP_CRED_MAPPER,
- NewRow: common.Row{
- "tenant_id": &common.ColumnVal{Value: "tenantid"},
- "appcred_id": &common.ColumnVal{Value: "aci"},
- "app_id": &common.ColumnVal{Value: "ai"},
- "apiprdt_id": &common.ColumnVal{Value: "api"},
- "status": &common.ColumnVal{Value: "s"},
- "_change_selector": &common.ColumnVal{Value: "c"},
- },
- },
- },
- }
-
- processChange(&insert)
- key := getKeyForDeveloperInfoCache("tenantid", "aci")
- developerInfo := developerInfoCache[key]
-
- Expect(developerInfo.ApiProduct).To(Equal("testproduct"))
- Expect(developerInfo.Developer).To(Equal("testdeveloper"))
- Expect(developerInfo.DeveloperEmail).To(Equal("testdeveloper@test.com"))
- Expect(developerInfo.DeveloperApp).To(Equal("name"))
})
})
})
