Merge pull request #16 from 30x/xapid_1064
[XAPID-1064] Add new API to apidAnalytics Plugin to support org/env
diff --git a/README.md b/README.md
index 699a2f6..da829f6 100644
--- a/README.md
+++ b/README.md
@@ -23,12 +23,11 @@
2. Create a listener for Apigee-Sync event
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
+3. Initialize POST /analytics/{scope_uuid} and POST /analytics API's
+4. Upon receiving requests
+ 1. Validate and enrich each batch of analytics records. If scope_uuid is given, then that is used to validate.
+ If scope_uuid is not provided, then the payload should have organization and environment. The org/env
+ is then used to validate the scope for this cluster.
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
@@ -53,5 +52,7 @@
### Exposed API
```sh
POST /analytics/{bundle_scope_uuid}
+POST /analytics
+
```
Complete spec is listed in `api.yaml`
diff --git a/api.go b/api.go
index 760407c..b93d327 100644
--- a/api.go
+++ b/api.go
@@ -37,6 +37,8 @@
analyticsBasePath = config.GetString(configAnalyticsBasePath)
services.API().HandleFunc(analyticsBasePath+"/{bundle_scope_uuid}",
saveAnalyticsRecord).Methods("POST")
+ services.API().HandleFunc(analyticsBasePath,
+ processAnalyticsRecord).Methods("POST")
}
func saveAnalyticsRecord(w http.ResponseWriter, r *http.Request) {
@@ -70,12 +72,63 @@
"UNKNOWN_SCOPE", dbErr.Reason)
}
} else {
- err := processPayload(tenant, scopeuuid, r)
+ body, err := getJsonBody(r)
if err.ErrorCode == "" {
- w.WriteHeader(http.StatusOK)
+ err = validateEnrichPublish(tenant, body)
+ if err.ErrorCode == "" {
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+ }
+ writeError(w, http.StatusBadRequest, err.ErrorCode, err.Reason)
+ }
+}
+
+func processAnalyticsRecord(w http.ResponseWriter, r *http.Request) {
+
+ w.Header().Set("Content-Type", "application/json; charset=UTF-8")
+
+ db := getDB() // When database isnt initialized
+ if db == nil {
+ writeError(w, http.StatusInternalServerError,
+ "INTERNAL_SERVER_ERROR",
+ "Service is not initialized completely")
+ return
+ }
+
+ if !strings.EqualFold(r.Header.Get("Content-Type"), "application/json") {
+ writeError(w, http.StatusBadRequest, "UNSUPPORTED_CONTENT_TYPE",
+ "Only supported content type is application/json")
+ return
+ }
+
+ body, err := getJsonBody(r)
+ if err.ErrorCode == "" {
+ tenant, e := getTenantFromPayload(body)
+ if e.ErrorCode == "" {
+ _, dbErr := validateTenant(tenant)
+ if dbErr.ErrorCode != "" {
+ switch dbErr.ErrorCode {
+ case "INTERNAL_SEARCH_ERROR":
+ writeError(w, http.StatusInternalServerError,
+ "INTERNAL_SEARCH_ERROR", dbErr.Reason)
+ case "UNKNOWN_SCOPE":
+ writeError(w, http.StatusBadRequest,
+ "UNKNOWN_SCOPE", dbErr.Reason)
+ }
+ return
+ } else {
+ err = validateEnrichPublish(tenant, body)
+ if err.ErrorCode == "" {
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+ }
} else {
writeError(w, http.StatusBadRequest,
- err.ErrorCode, err.Reason)
+ e.ErrorCode, e.Reason)
+ return
}
}
+ writeError(w, http.StatusBadRequest, err.ErrorCode, err.Reason)
}
diff --git a/api.yaml b/api.yaml
index 1ea4597..0414d5b 100644
--- a/api.yaml
+++ b/api.yaml
@@ -26,6 +26,32 @@
produces:
- application/json
paths:
+ '/analytics':
+ x-swagger-router-controller: analytics
+ parameters:
+ - name: analytics_data
+ in: body
+ description: The analytics data you want to post
+ required: true
+ schema:
+ $ref: "#/definitions/analytics_data"
+ post:
+ responses:
+ "200":
+ description: Success
+ "400":
+ description: Bad Request
+ schema:
+ $ref: "#/definitions/errClientError"
+ "500":
+ description: Server error
+ schema:
+ $ref: "#/definitions/errServerError"
+ default:
+ description: Error
+ schema:
+ $ref: "#/definitions/errResponse"
+
'/analytics/{bundle_scope_uuid}':
x-swagger-router-controller: analytics
parameters:
@@ -34,7 +60,7 @@
required: true
description: bundle UUID that can be mapped to a scope by APID
type: string
- - name: analytics_data
+ - name: records
in: body
description: The analytics data you want to post
required: true
@@ -58,6 +84,53 @@
$ref: "#/definitions/errResponse"
definitions:
+ analytics_data:
+ type: object
+ required:
+ - organization
+ - environment
+ - records
+ properties:
+ organization:
+ type: string
+ environment:
+ type: string
+ records:
+ type: object
+ $ref: "#/definitions/records"
+ example: {
+ "organization":"orgname",
+ "enironment":"envname",
+ "records":[{
+ "response_status_code": 400,
+ "client_received_start_timestamp": 1462850097576,
+ "client_received_end_timestamp": 1462850097580,
+ "client_id":"0GJKn7EQmNkKYcGL7x3gHaawWLs5gUPr"
+ },{
+ "response_status_code": 200,
+ "client_id":"2ngXgr6Rl2PXWiEmbt8zCkWY3Ptjb8ep",
+ "request_verb" : "GET",
+ "api_product":" test_product",
+ "access_token" : "fewGWG343LDV346345YCDS",
+ "apiproxy" : "OAuthProxy",
+ "apiproxy_revision" : "2",
+ "client_ip": "10.16.9.11",
+ "client_sent_end_timestamp": 1462850097894,
+ "client_received_start_timestamp": 1462850097576,
+ "client_received_end_timestamp": 1462850097580,
+ "client_sent_start_timestamp": 1462850097894,
+ "request_path" : "/oauth/oauthv2/auth_code/",
+ "request_uri": "/oauth/oauthv2/auth_code/?response_type=code&redirect_url=http%3A%2F%2Fexample.com&client_id=A1h6yYAVeADnEKji8M37zCSn6olcmQDB",
+ "useragent" : "Chrome",
+ "target" : "target_name",
+ "target_received_end_timestamp": 1462850097800,
+ "target_received_start_timestamp": 1462850097800,
+ "target_response_code" : 200,
+ "target_sent_end_timestamp" : 1462850097802,
+ "target_sent_start_timestamp" : 1462850097802
+ }]
+ }
+
records:
type: object
required:
@@ -72,6 +145,7 @@
"records":[{
"response_status_code": 400,
"client_received_start_timestamp": 1462850097576,
+ "client_received_end_timestamp": 1462850097580,
"client_id":"0GJKn7EQmNkKYcGL7x3gHaawWLs5gUPr"
},{
"response_status_code": 200,
@@ -103,13 +177,18 @@
type: object
required:
- client_received_start_timestamp
+ - client_received_end_timestamp
properties:
client_received_start_timestamp:
type: integer
format: int64
+ client_received_end_timestamp:
+ type: integer
+ format: int64
example: {
"response_status_code":400,
"client_received_start_timestamp":1462850097576,
+ "client_received_end_timestamp": 1462850097580,
"client_id":"0GJKn7EQmNkKYcGL7x3gHaawWLs5gUPr"
}
diff --git a/api_helper.go b/api_helper.go
index 3b6f63d..67e9725 100644
--- a/api_helper.go
+++ b/api_helper.go
@@ -41,16 +41,15 @@
}
type tenant struct {
- Org string
- Env string
- TenantId string
+ Org string
+ Env string
}
-func processPayload(tenant tenant, scopeuuid string, r *http.Request) errResponse {
+func getJsonBody(r *http.Request) (map[string]interface{}, errResponse) {
var gzipEncoded bool
if r.Header.Get("Content-Encoding") != "" {
if !strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") {
- return errResponse{
+ return nil, errResponse{
ErrorCode: "UNSUPPORTED_CONTENT_ENCODING",
Reason: "Only supported content encoding is gzip"}
} else {
@@ -63,7 +62,7 @@
if gzipEncoded {
reader, err = gzip.NewReader(r.Body) // reader for gzip encoded data
if err != nil {
- return errResponse{
+ return nil, errResponse{
ErrorCode: "BAD_DATA",
Reason: "Gzip Encoded data cannot be read"}
}
@@ -71,30 +70,60 @@
reader = r.Body
}
- errMessage := validateEnrichPublish(tenant, scopeuuid, reader)
- if errMessage.ErrorCode != "" {
- return errMessage
- }
- return errResponse{}
-}
-
-func validateEnrichPublish(tenant tenant, scopeuuid string, reader io.Reader) errResponse {
var raw map[string]interface{}
decoder := json.NewDecoder(reader) // Decode payload to JSON data
decoder.UseNumber()
if err := decoder.Decode(&raw); err != nil {
- return errResponse{ErrorCode: "BAD_DATA",
+ return nil, errResponse{ErrorCode: "BAD_DATA",
Reason: "Not a valid JSON payload"}
}
+ return raw, errResponse{}
+}
+
+/*
+Get tenant from payload based on the 2 required fields - organization and environment
+*/
+func getTenantFromPayload(raw map[string]interface{}) (tenant, errResponse) {
+ elems := []string{"organization", "environment"}
+ for _, elem := range elems {
+ if raw[elem] == nil || raw[elem].(string) == "" {
+ return tenant{}, errResponse{
+ ErrorCode: "MISSING_FIELD",
+ Reason: "Missing Required field: " + elem}
+ }
+ }
+
+ org := raw["organization"].(string)
+ env := raw["environment"].(string)
+ return tenant{Org: org, Env: env}, errResponse{}
+}
+
+func validateEnrichPublish(tenant tenant, raw map[string]interface{}) errResponse {
if records := raw["records"]; records != nil {
+ records, isArray := records.([]interface{})
+ if !isArray {
+ return errResponse{
+ ErrorCode: "BAD_DATA",
+ Reason: "records should be a list of analytics records"}
+ }
+ if len(records) == 0 {
+ return errResponse{
+ ErrorCode: "NO_RECORDS",
+ Reason: "No analytics records in the payload"}
+ }
// Iterate through each record to validate and enrich it
- for _, eachRecord := range records.([]interface{}) {
- recordMap := eachRecord.(map[string]interface{})
+ for _, eachRecord := range records {
+ recordMap, isMap := eachRecord.(map[string]interface{})
+ if !isMap {
+ return errResponse{
+ ErrorCode: "BAD_DATA",
+ Reason: "Each Analytics record in records should be a json object"}
+ }
valid, err := validate(recordMap)
if valid {
- enrich(recordMap, scopeuuid, tenant)
+ enrich(recordMap, tenant)
} else {
// Even if there is one bad record, then reject entire batch
return err
@@ -102,7 +131,7 @@
}
axRecords := axRecords{
Tenant: tenant,
- Records: records.([]interface{})}
+ Records: records}
// publish batch of records to channel (blocking call)
internalBuffer <- axRecords
} else {
@@ -116,7 +145,8 @@
/*
Does basic validation on each analytics message
1. client_received_start_timestamp, client_received_end_timestamp should exist
-2. client_received_end_timestamp should be > client_received_start_timestamp and not 0
+2. client_received_start_timestamp, client_received_end_timestamp should be a number
+3. client_received_end_timestamp should be > client_received_start_timestamp and not 0
*/
func validate(recordMap map[string]interface{}) (bool, errResponse) {
elems := []string{"client_received_start_timestamp", "client_received_end_timestamp"}
@@ -131,12 +161,19 @@
crst, exists1 := recordMap["client_received_start_timestamp"]
cret, exists2 := recordMap["client_received_end_timestamp"]
if exists1 && exists2 {
- if crst.(json.Number) == json.Number("0") || cret.(json.Number) == json.Number("0") {
+ crst, isNumber1 := crst.(json.Number)
+ cret, isNumber2 := cret.(json.Number)
+ if !isNumber1 || !isNumber2 {
+ return false, errResponse{
+ ErrorCode: "BAD_DATA",
+ Reason: "client_received_start_timestamp and " +
+ "client_received_end_timestamp has to be number"}
+ } else if crst == json.Number("0") || cret == json.Number("0") {
return false, errResponse{
ErrorCode: "BAD_DATA",
Reason: "client_received_start_timestamp or " +
- "> client_received_end_timestamp cannot be 0"}
- } else if crst.(json.Number) > cret.(json.Number) {
+ "client_received_end_timestamp cannot be 0"}
+ } else if crst > cret {
return false, errResponse{
ErrorCode: "BAD_DATA",
Reason: "client_received_start_timestamp " +
@@ -148,43 +185,11 @@
/*
Enrich each record by adding org and env fields
-It also finds add developer related information based on the apiKey
*/
-func enrich(recordMap map[string]interface{}, scopeuuid string, tenant tenant) {
- org, orgExists := recordMap["organization"]
- if !orgExists || org.(string) == "" {
- recordMap["organization"] = tenant.Org
- }
-
- env, envExists := recordMap["environment"]
- if !envExists || env.(string) == "" {
- recordMap["environment"] = tenant.Env
- }
-
- apiKey, exists := recordMap["client_id"]
- // apiKey doesnt exist then ignore adding developer fields
- if exists {
- apiKey := apiKey.(string)
- if apiKey != "" {
- devInfo := getDeveloperInfo(tenant.TenantId, apiKey)
- _, exists := recordMap["api_product"]
- if !exists {
- recordMap["api_product"] = devInfo.ApiProduct
- }
- _, exists = recordMap["developer_app"]
- if !exists {
- recordMap["developer_app"] = devInfo.DeveloperApp
- }
- _, exists = recordMap["developer_email"]
- if !exists {
- recordMap["developer_email"] = devInfo.DeveloperEmail
- }
- _, exists = recordMap["developer"]
- if !exists {
- recordMap["developer"] = devInfo.Developer
- }
- }
- }
+func enrich(recordMap map[string]interface{}, tenant tenant) {
+ // Always overwrite organization/environment value with the tenant information provided in the payload
+ recordMap["organization"] = tenant.Org
+ recordMap["environment"] = tenant.Env
}
func writeError(w http.ResponseWriter, status int, code string, reason string) {
diff --git a/api_helper_test.go b/api_helper_test.go
index 2f96267..4101c5c 100644
--- a/api_helper_test.go
+++ b/api_helper_test.go
@@ -23,6 +23,38 @@
)
// BeforeSuite setup and AfterSuite cleanup is in apidAnalytics_suite_test.go
+var _ = Describe("test getTenantFromPayload()", func() {
+ Context("invalid record", func() {
+ It("should return invalid record", func() {
+ By("payload with missing required keys")
+
+ var payload = []byte(`{
+ "records":[{
+ "response_status_code": 200,
+ "client_id":"testapikey"
+ }]}`)
+ raw := getRaw(payload)
+ _, e := getTenantFromPayload(raw)
+ Expect(e.ErrorCode).To(Equal("MISSING_FIELD"))
+ })
+ })
+ Context("valid record", func() {
+ It("should return tenant with org and env", func() {
+ var payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":[{
+ "response_status_code": 200,
+ "client_id":"testapikey"
+ }]}`)
+ raw := getRaw(payload)
+ tenant, _ := getTenantFromPayload(raw)
+ Expect(tenant.Org).To(Equal("testorg"))
+ Expect(tenant.Env).To(Equal("testenv"))
+ })
+ })
+})
+
var _ = Describe("test valid() directly", func() {
Context("invalid record", func() {
It("should return invalid record", func() {
@@ -66,7 +98,35 @@
Expect(valid).To(BeFalse())
Expect(e.ErrorCode).To(Equal("BAD_DATA"))
Expect(e.Reason).To(Equal("client_received_start_timestamp or " +
- "> client_received_end_timestamp cannot be 0"))
+ "client_received_end_timestamp cannot be 0"))
+
+ By("payload with clst = null")
+ record = []byte(`{
+ "response_status_code": 200,
+ "client_id":"testapikey",
+ "client_received_start_timestamp": null,
+ "client_received_end_timestamp": 1486406248260
+ }`)
+ raw = getRaw(record)
+ valid, e = validate(raw)
+
+ Expect(valid).To(BeFalse())
+ Expect(e.ErrorCode).To(Equal("MISSING_FIELD"))
+
+ By("payload with clst as a string")
+ record = []byte(`{
+ "response_status_code": 200,
+ "client_id":"testapikey",
+ "client_received_start_timestamp": "",
+ "client_received_end_timestamp": 1486406248260
+ }`)
+ raw = getRaw(record)
+ valid, e = validate(raw)
+
+ Expect(valid).To(BeFalse())
+ Expect(e.ErrorCode).To(Equal("BAD_DATA"))
+ Expect(e.Reason).To(Equal("client_received_start_timestamp and " +
+ "client_received_end_timestamp has to be number"))
})
})
Context("valid record", func() {
@@ -85,46 +145,55 @@
})
var _ = Describe("test enrich() directly", func() {
- Context("enrich record for existing apiKey", func() {
- It("developer related fields should be added", func() {
+ Context("enrich record where org/env in record is different from main org/env in payload", func() {
+ It("The record should also have org/env for which record was validated ", func() {
+ var record = []byte(`{
+ "organization":"o",
+ "environment":"e",
+ "client_id":"testapikey",
+ "client_received_start_timestamp": 1486406248277,
+ "client_received_end_timestamp": 1486406248290
+ }`)
+
+ raw := getRaw(record)
+ tenant := tenant{Org: "testorg", Env: "testenv"}
+ enrich(raw, tenant)
+
+ Expect(raw["organization"]).To(Equal(tenant.Org))
+ Expect(raw["environment"]).To(Equal(tenant.Env))
+ })
+ })
+ Context("enrich record where no org/env is there in the record is set", func() {
+ It("developer related fields should not be added", func() {
var record = []byte(`{
"response_status_code": 200,
"client_id":"testapikey",
"client_received_start_timestamp": 1486406248277,
"client_received_end_timestamp": 1486406248290
}`)
-
raw := getRaw(record)
- tenant := tenant{Org: "testorg", Env: "testenv", TenantId: "tenantid"}
- enrich(raw, "testid", tenant)
+ tenant := tenant{Org: "testorg", Env: "testenv"}
+ enrich(raw, tenant)
Expect(raw["organization"]).To(Equal(tenant.Org))
Expect(raw["environment"]).To(Equal(tenant.Env))
- Expect(raw["api_product"]).To(Equal("testproduct"))
- Expect(raw["developer_app"]).To(Equal("testapp"))
- Expect(raw["developer_email"]).To(Equal("testdeveloper@test.com"))
- Expect(raw["developer"]).To(Equal("testdeveloper"))
})
})
-
- Context("enrich record where no apikey is set", func() {
+ Context("enrich record where org/env is same as the main org/env in payload", func() {
It("developer related fields should not be added", func() {
var record = []byte(`{
- "response_status_code": 200,
+ "organization":"testorg",
+ "environment": "testenv",
+ "client_id":"testapikey",
"client_received_start_timestamp": 1486406248277,
"client_received_end_timestamp": 1486406248290
- }`)
-
+ }`)
raw := getRaw(record)
- tenant := tenant{Org: "testorg", Env: "testenv", TenantId: "tenantid"}
- enrich(raw, "testid", tenant)
+ tenant := tenant{Org: "testorg", Env: "testenv"}
+ enrich(raw, tenant)
Expect(raw["organization"]).To(Equal(tenant.Org))
Expect(raw["environment"]).To(Equal(tenant.Env))
- Expect(raw["api_product"]).To(BeNil())
- Expect(raw["developer_app"]).To(BeNil())
- Expect(raw["developer_email"]).To(BeNil())
- Expect(raw["developer"]).To(BeNil())
})
})
})
diff --git a/api_test.go b/api_test.go
index 1c253c6..96664c8 100644
--- a/api_test.go
+++ b/api_test.go
@@ -30,15 +30,10 @@
var _ = Describe("POST /analytics/{scopeuuid}", func() {
Context("invalid content type header", func() {
It("should return bad request", func() {
- uri, err := url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "test")
- Expect(err).ShouldNot(HaveOccurred())
-
- req, _ := http.NewRequest("POST", uri.String(), nil)
+ req := getRequestWithScope("test", nil)
req.Header.Set("Content-Type", "application/x-gzip")
res, e := makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("UNSUPPORTED_CONTENT_TYPE"))
})
@@ -46,16 +41,10 @@
Context("invalid content encoding header", func() {
It("should return bad request", func() {
- uri, err := url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
- req, _ := http.NewRequest("POST", uri.String(), nil)
- req.Header.Set("Content-Type", "application/json")
+ req := getRequestWithScope("testid", nil)
req.Header.Set("Content-Encoding", "application/gzip")
res, e := makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("UNSUPPORTED_CONTENT_ENCODING"))
})
@@ -63,15 +52,9 @@
Context("invalid scopeuuid", func() {
It("should return bad request", func() {
- uri, err := url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "wrongid")
- Expect(err).ShouldNot(HaveOccurred())
-
- req, _ := http.NewRequest("POST", uri.String(), nil)
- req.Header.Set("Content-Type", "application/json")
+ req := getRequestWithScope("wrongid", nil)
res, e := makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
})
@@ -82,20 +65,13 @@
db := getDB()
setDB(nil)
- uri, err := url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
- req, _ := http.NewRequest("POST", uri.String(), nil)
- req.Header.Set("Content-Type", "application/json")
+ req := getRequestWithScope("testid", nil)
res, e := makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusInternalServerError))
Expect(e.ErrorCode).To(Equal("INTERNAL_SERVER_ERROR"))
setDB(db)
-
})
})
@@ -103,60 +79,34 @@
It("should return bad request", func() {
By("no payload")
- uri, err := url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
- req, _ := http.NewRequest("POST", uri.String(), nil)
- req.Header.Set("Content-Type", "application/json")
-
+ req := getRequestWithScope("testid", nil)
res, e := makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("BAD_DATA"))
Expect(e.Reason).To(Equal("Not a valid JSON payload"))
By("payload with 0 records")
- uri, err = url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
var payload = []byte(`{}`)
- req, _ = http.NewRequest("POST", uri.String(), bytes.NewReader(payload))
- req.Header.Set("Content-Type", "application/json")
-
+ req = getRequestWithScope("testid", payload)
res, e = makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("NO_RECORDS"))
By("set content encoding to gzip but send json data")
- uri, err = url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
payload = []byte(`{
"records":[{
"response_status_code": 200,
"client_id":"testapikey"
}]
}`)
-
- req, _ = http.NewRequest("POST", uri.String(), bytes.NewReader(payload))
- req.Header.Set("Content-Type", "application/json")
+ req = getRequestWithScope("testid", payload)
req.Header.Set("Content-Encoding", "gzip")
-
res, e = makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("BAD_DATA"))
Expect(e.Reason).To(Equal("Gzip Encoded data cannot be read"))
By("1 bad record")
- uri, err = url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
payload = []byte(`{
"records":[{
"response_status_code": 200,
@@ -168,12 +118,8 @@
"client_id":"testapikey"
}]
}`)
-
- req, _ = http.NewRequest("POST", uri.String(), bytes.NewReader(payload))
- req.Header.Set("Content-Type", "application/json")
-
+ req = getRequestWithScope("testid", payload)
res, e = makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
Expect(e.ErrorCode).To(Equal("MISSING_FIELD"))
})
@@ -181,10 +127,6 @@
Context("valid payload", func() {
It("should return successfully", func() {
- uri, err := url.Parse(testServer.URL)
- uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", "testid")
- Expect(err).ShouldNot(HaveOccurred())
-
var payload = []byte(`{
"records":[{
"response_status_code": 200,
@@ -193,17 +135,195 @@
"client_received_end_timestamp": 1486406248290
}]
}`)
-
- req, _ := http.NewRequest("POST", uri.String(), bytes.NewReader(payload))
- req.Header.Set("Content-Type", "application/json")
-
+ req := getRequestWithScope("testid", payload)
res, _ := makeRequest(req)
-
Expect(res.StatusCode).To(Equal(http.StatusOK))
})
})
})
+var _ = Describe("POST /analytics", func() {
+ Context("invalid content type header", func() {
+ It("should return bad request", func() {
+ req := getRequest(nil)
+ req.Header.Set("Content-Type", "application/x-gzip")
+ res, e := makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("UNSUPPORTED_CONTENT_TYPE"))
+ })
+ })
+
+ Context("invalid content encoding header", func() {
+ It("should return bad request", func() {
+ req := getRequest(nil)
+ req.Header.Set("Content-Encoding", "application/gzip")
+ res, e := makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("UNSUPPORTED_CONTENT_ENCODING"))
+ })
+ })
+
+ Context("Unitialized DB", func() {
+ It("should return internal server error", func() {
+ db := getDB()
+ setDB(nil)
+
+ req := getRequest(nil)
+ res, e := makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusInternalServerError))
+ Expect(e.ErrorCode).To(Equal("INTERNAL_SERVER_ERROR"))
+
+ setDB(db)
+ })
+ })
+
+ Context("bad payload", func() {
+ It("should return bad request", func() {
+
+ By("no payload")
+ req := getRequest(nil)
+ res, e := makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("BAD_DATA"))
+ Expect(e.Reason).To(Equal("Not a valid JSON payload"))
+
+ By("payload with no organization and environment")
+ var payload = []byte(`{}`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("MISSING_FIELD"))
+
+ By("payload with wrong organization and environment")
+ payload = []byte(`{
+ "organization":"wrongorg",
+ "environment":"wrongenv",
+ "records":[{
+ "response_status_code": 200,
+ "client_id":"testapikey"
+ }]
+ }`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
+
+ By("payload with no records")
+ payload = []byte(`{"organization":"testorg","environment":"testenv"}`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("NO_RECORDS"))
+
+ By("payload with 0 records")
+ payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":[]
+ }`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("NO_RECORDS"))
+
+ By("payload with records not as list")
+ payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":{}
+ }`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("BAD_DATA"))
+
+ By("payload with each record not a json in an array")
+ payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":[""]
+ }`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("BAD_DATA"))
+
+ By("set content encoding to gzip but send json data")
+ payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":[{
+ "response_status_code": 200,
+ "client_id":"testapikey"
+ }]
+ }`)
+ req = getRequest(payload)
+ req.Header.Set("Content-Encoding", "gzip")
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("BAD_DATA"))
+ Expect(e.Reason).To(Equal("Gzip Encoded data cannot be read"))
+
+ By("1 bad record")
+ payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":[{
+ "response_status_code": 200,
+ "client_id":"testapikey",
+ "client_received_start_timestamp": 1486406248277,
+ "client_received_end_timestamp": 1486406248290
+ },{
+ "response_status_code": 200,
+ "client_id":"testapikey"
+ }]
+ }`)
+ req = getRequest(payload)
+ res, e = makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusBadRequest))
+ Expect(e.ErrorCode).To(Equal("MISSING_FIELD"))
+ })
+ })
+
+ Context("valid payload", func() {
+ It("should return successfully", func() {
+ var payload = []byte(`{
+ "organization":"testorg",
+ "environment":"testenv",
+ "records":[{
+ "response_status_code": 200,
+ "client_id":"testapikey",
+ "client_received_start_timestamp": 1486406248277,
+ "client_received_end_timestamp": 1486406248290
+ }]
+ }`)
+ req := getRequest(payload)
+ res, _ := makeRequest(req)
+ Expect(res.StatusCode).To(Equal(http.StatusOK))
+ })
+ })
+})
+
+func getRequest(payload []byte) *http.Request {
+ uri, err := url.Parse(testServer.URL)
+ uri.Path = analyticsBasePath
+ Expect(err).ShouldNot(HaveOccurred())
+
+ req, _ := http.NewRequest("POST", uri.String(), bytes.NewReader(payload))
+ req.Header.Set("Content-Type", "application/json")
+ return req
+}
+
+func getRequestWithScope(scope string, payload []byte) *http.Request {
+ uri, err := url.Parse(testServer.URL)
+ uri.Path = fmt.Sprintf(analyticsBasePath+"/%s", scope)
+ Expect(err).ShouldNot(HaveOccurred())
+
+ req, _ := http.NewRequest("POST", uri.String(), bytes.NewReader(payload))
+ req.Header.Set("Content-Type", "application/json")
+ return req
+}
+
func makeRequest(req *http.Request) (*http.Response, errResponse) {
res, err := client.Do(req)
defer res.Body.Close()
@@ -212,6 +332,5 @@
var body errResponse
respBody, _ := ioutil.ReadAll(res.Body)
json.Unmarshal(respBody, &body)
-
return res, body
}
diff --git a/buffering_manager_test.go b/buffering_manager_test.go
index 2b2fbd6..caf50c5 100644
--- a/buffering_manager_test.go
+++ b/buffering_manager_test.go
@@ -27,7 +27,7 @@
var _ = Describe("test getBucketForTimestamp()", func() {
It("should return new bucket or existing bucket if created previously", func() {
t := time.Date(2017, 1, 20, 10, 24, 5, 0, time.UTC)
- tenant := tenant{Org: "testorg", Env: "testenv", TenantId: "tenantid"}
+ tenant := tenant{Org: "testorg", Env: "testenv"}
bucket, err := getBucketForTimestamp(t, tenant)
Expect(err).ShouldNot(HaveOccurred())
diff --git a/common_helper.go b/common_helper.go
index 813acc4..05805bf 100644
--- a/common_helper.go
+++ b/common_helper.go
@@ -16,7 +16,6 @@
import (
"database/sql"
- "github.com/apigee-labs/transicator/common"
"sync"
)
@@ -27,54 +26,69 @@
// read while its being written to and vice versa
var tenantCachelock = sync.RWMutex{}
-// Cache for apiKey~tenantId to developer related information
-var developerInfoCache map[string]developerInfo
+// Cache for all org/env for this cluster
+var orgEnvCache map[string]bool
-// RW lock for developerInfo map cache since the cache can be
+// RW lock for orgEnvCache map cache since the cache can be
// read while its being written to and vice versa
-var developerInfoCacheLock = sync.RWMutex{}
+var orgEnvCacheLock = sync.RWMutex{}
// Load data scope information into an in-memory cache so that
// for each record a DB lookup is not required
-func createTenantCache(snapshot *common.Snapshot) {
+func createTenantCache() {
// Lock before writing to the map as it has multiple readers
tenantCachelock.Lock()
defer tenantCachelock.Unlock()
tenantCache = make(map[string]tenant)
- for _, table := range snapshot.Tables {
- switch table.Name {
- case "edgex.data_scope":
- for _, row := range table.Rows {
- var org, env, tenantId, id string
+ var org, env, id string
- row.Get("id", &id)
- row.Get("scope", &tenantId)
- row.Get("org", &org)
- row.Get("env", &env)
- if id != "" {
- tenantCache[id] = tenant{Org: org,
- Env: env,
- TenantId: tenantId}
- }
- }
+ db := getDB()
+ rows, error := db.Query("SELECT env, org, id FROM edgex_data_scope")
+
+ if error != nil {
+ log.Warnf("Could not get datascope from DB due to : %s", error.Error())
+ } else {
+ defer rows.Close()
+ // Lock before writing to the map as it has multiple readers
+ for rows.Next() {
+ rows.Scan(&env, &org, &id)
+ tenantCache[id] = tenant{Org: org, Env: env}
}
}
+
log.Debugf("Count of data scopes in the cache: %d", len(tenantCache))
}
// Load data scope information into an in-memory cache so that
// for each record a DB lookup is not required
-func updateDeveloperInfoCache() {
+func createOrgEnvCache() {
// Lock before writing to the map as it has multiple readers
- developerInfoCacheLock.Lock()
- defer developerInfoCacheLock.Unlock()
- developerInfoCache = make(map[string]developerInfo)
- log.Debug("Invalidated developerInfo cache")
+ orgEnvCacheLock.Lock()
+ defer orgEnvCacheLock.Unlock()
+ orgEnvCache = make(map[string]bool)
+
+ var org, env string
+ db := getDB()
+
+ rows, error := db.Query("SELECT env, org FROM edgex_data_scope")
+
+ if error != nil {
+ log.Warnf("Could not get datascope from DB due to : %s", error.Error())
+ } else {
+ defer rows.Close()
+ // Lock before writing to the map as it has multiple readers
+ for rows.Next() {
+ rows.Scan(&env, &org)
+ orgEnv := getKeyForOrgEnvCache(org, env)
+ orgEnvCache[orgEnv] = true
+ }
+ }
+ log.Debugf("Count of org~env in the cache: %d", len(orgEnvCache))
}
// Returns Tenant Info given a scope uuid from the cache or by querying
-// the DB directly based on useCachig config
+// the DB directly based on useCaching config
func getTenantForScope(scopeuuid string) (tenant, dbError) {
if config.GetBool(useCaching) {
// acquire a read lock as this cache has 1 writer as well
@@ -85,8 +99,7 @@
if !exists {
log.Debugf("No tenant found for scopeuuid = %s "+
- "in cache", scopeuuid)
- log.Debug("loading info from DB")
+ "in cache, loading info from DB", scopeuuid)
// Update cache
t, err := getTenantFromDB(scopeuuid)
@@ -108,49 +121,13 @@
}
}
-// 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
- developerInfoCacheLock.RLock()
- devInfo, exists := developerInfoCache[keyForMap]
- developerInfoCacheLock.RUnlock()
-
- if !exists {
- 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 {
- devInfo, _ := getDevInfoFromDB(tenantId, apiKey)
- return devInfo
- }
-}
-
// Returns tenant info by querying DB directly
func getTenantFromDB(scopeuuid string) (tenant, dbError) {
- var org, env, tenantId string
+ var org, env string
db := getDB()
- error := db.QueryRow("SELECT env, org, scope FROM edgex_data_scope"+
- " where id = ?", scopeuuid).Scan(&env, &org, &tenantId)
+ error := db.QueryRow("SELECT env, org FROM edgex_data_scope"+
+ " where id = ?", scopeuuid).Scan(&env, &org)
switch {
case error == sql.ErrNoRows:
@@ -167,59 +144,67 @@
Reason: reason}
}
return tenant{
- Org: org,
- Env: env,
- TenantId: tenantId}, dbError{}
+ Org: org,
+ Env: env}, 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 kms_app_credential_apiproduct_mapper as mp " +
- "INNER JOIN kms_api_product as ap ON ap.id = mp.apiprdt_id " +
- "INNER JOIN kms_app AS a ON a.id = mp.app_id " +
- "INNER JOIN kms_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 {
- return sqlValue.String
+/*
+Checks if given org/env exists is a valid scope for this apid cluster
+It also stores the scope i.e. tenant_id in the tenant object using pointer.
+tenant_id in combination with apiKey is used to find kms related information
+*/
+func validateTenant(tenant tenant) (bool, dbError) {
+ if config.GetBool(useCaching) {
+ // acquire a read lock as this cache has 1 writer as well
+ orgEnvCacheLock.RLock()
+ orgEnv := getKeyForOrgEnvCache(tenant.Org, tenant.Env)
+ _, exists := orgEnvCache[orgEnv]
+ orgEnvCacheLock.RUnlock()
+ dbErr := dbError{}
+ if !exists {
+ log.Debugf("OrgEnv = %s not found "+
+ "in cache, loading info from DB", orgEnv)
+ // Update cache
+ valid, dbErr := validateTenantFromDB(tenant)
+ if valid {
+ // update cache
+ orgEnvCacheLock.Lock()
+ defer orgEnvCacheLock.Unlock()
+ orgEnvCache[orgEnv] = true
+ }
+ return valid, dbErr
+ } else {
+ return true, dbErr
+ }
} else {
- return ""
+ return validateTenantFromDB(tenant)
}
+
}
-// Build Key as a combination of tenantId and apiKey for the developerInfo Cache
-func getKeyForDeveloperInfoCache(tenantId string, apiKey string) string {
- return tenantId + "~" + apiKey
+func validateTenantFromDB(tenant tenant) (bool, dbError) {
+ db := getDB()
+ rows, err := db.Query("SELECT 1 FROM edgex_data_scope"+
+ " where org = ? and env = ?", tenant.Org, tenant.Env)
+
+ if !rows.Next() {
+ if err == nil {
+ reason := "No tenant found for this org: " + tenant.Org + " and env:" + tenant.Env
+ errorCode := "UNKNOWN_SCOPE"
+ return false, dbError{
+ ErrorCode: errorCode,
+ Reason: reason}
+ } else {
+ reason := err.Error()
+ errorCode := "INTERNAL_SEARCH_ERROR"
+ return false, dbError{
+ ErrorCode: errorCode,
+ Reason: reason}
+ }
+ }
+ return true, dbError{}
+}
+
+func getKeyForOrgEnvCache(org, env string) string {
+ return org + "~" + env
}
diff --git a/common_helper_test.go b/common_helper_test.go
index 901d7e2..c4e8e90 100644
--- a/common_helper_test.go
+++ b/common_helper_test.go
@@ -15,19 +15,29 @@
package apidAnalytics
import (
- "database/sql"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
-
- "github.com/apigee-labs/transicator/common"
)
+var _ = Describe("test createTenantCache()", func() {
+ It("It should create a cache from DB", func() {
+ createTenantCache()
+ Expect(len(tenantCache)).To(Equal(1))
+ })
+})
+
+var _ = Describe("test createOrgEnvCache()", func() {
+ It("It should create a cache from DB", func() {
+ createOrgEnvCache()
+ Expect(len(orgEnvCache)).To(Equal(1))
+ })
+})
+
var _ = Describe("test getTenantForScope()", func() {
Context("with usecaching set to true", func() {
BeforeEach(func() {
config.Set(useCaching, true)
- snapshot := getDatascopeSnapshot()
- createTenantCache(&snapshot)
+ createTenantCache()
Expect(len(tenantCache)).To(Equal(1))
})
AfterEach(func() {
@@ -37,10 +47,9 @@
It("should return testorg and testenv", func() {
tenant, dbError := getTenantForScope("testid")
Expect(dbError.Reason).To(Equal(""))
- Expect(tenant.TenantId).To(Equal("tenantid"))
+ Expect(tenant.Org).To(Equal("testorg"))
})
})
-
Context("get tenant for invalid scopeuuid", func() {
It("should return empty tenant and a db error", func() {
tenant, dbError := getTenantForScope("wrongid")
@@ -58,7 +67,6 @@
Expect(tenant.Org).To(Equal("testorg"))
})
})
-
Context("get tenant for invalid scopeuuid", func() {
It("should return empty tenant and a db error", func() {
tenant, dbError := getTenantForScope("wrongid")
@@ -70,123 +78,96 @@
})
var _ = Describe("test getTenantFromDB()", func() {
- Context("get developerInfo for valid scopeuuid", func() {
- It("should return all right data", func() {
+ Context("get tenant for valid scopeuuid", func() {
+ It("should return testorg and testenv", func() {
tenant, dbError := getTenantFromDB("testid")
Expect(dbError.Reason).To(Equal(""))
- Expect(tenant.TenantId).To(Equal("tenantid"))
+ Expect(tenant.Org).To(Equal("testorg"))
+ Expect(tenant.Env).To(Equal("testenv"))
+
})
})
- Context("get developerInfo for invalid scopeuuid", func() {
- It("should return error", func() {
+ Context("get tenant for invalid scopeuuid", func() {
+ It("should return error with unknown scope", func() {
tenant, dbError := getTenantFromDB("wrongid")
Expect(tenant.Org).To(Equal(""))
Expect(dbError.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
})
})
+})
+
+var _ = Describe("test validateTenant()", func() {
+
+ Context("with usecaching set to true", func() {
+ BeforeEach(func() {
+ config.Set(useCaching, true)
+ createOrgEnvCache()
+ Expect(len(orgEnvCache)).To(Equal(1))
+ })
+ AfterEach(func() {
+ config.Set(useCaching, false)
+ })
+ Context("valididate existing org/env", func() {
+ It("should return true", func() {
+ tenant := tenant{Org: "testorg", Env: "testenv"}
+ valid, dbError := validateTenant(tenant)
+ Expect(dbError.Reason).To(Equal(""))
+ Expect(valid).To(BeTrue())
+ })
+ })
+
+ Context("get tenant for invalid scopeuuid", func() {
+ It("should return false", func() {
+ tenant := tenant{Org: "wrongorg", Env: "wrongenv"}
+ valid, dbError := validateTenant(tenant)
+ Expect(dbError.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
+ Expect(valid).To(BeFalse())
+ })
+ })
+ })
+ Context("with usecaching set to false", func() {
+ Context("valididate existing org/env", func() {
+ It("should return true", func() {
+ tenant := tenant{Org: "testorg", Env: "testenv"}
+ valid, dbError := validateTenant(tenant)
+ Expect(dbError.Reason).To(Equal(""))
+ Expect(valid).To(BeTrue())
+ })
+ })
+ Context("get tenant for invalid scopeuuid", func() {
+ It("should return false", func() {
+ tenant := tenant{Org: "wrongorg", Env: "wrongenv"}
+ valid, dbError := validateTenant(tenant)
+ Expect(dbError.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
+ Expect(valid).To(BeFalse())
+ })
+ })
+ })
+})
+
+var _ = Describe("test validateTenantFromDB()", func() {
+ Context("validate tenant for org/env that exists in DB", func() {
+ It("should not return an error and valid should be true", func() {
+ tenant := tenant{Org: "testorg", Env: "testenv"}
+ valid, dbError := validateTenantFromDB(tenant)
+ Expect(valid).To(BeTrue())
+ Expect(dbError.ErrorCode).To(Equal(""))
+ })
+ })
+ Context("validate tenant for org/env that do not exist in DB", func() {
+ It("should return error with unknown_scope", func() {
+ tenant := tenant{Org: "wrongorg", Env: "wrongenv"}
+ valid, dbError := validateTenantFromDB(tenant)
+ Expect(valid).To(BeFalse())
+ 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() {
- 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 empty", func() {
- developerInfo := getDeveloperInfo("wrongid", "wrongapikey")
- Expect(developerInfo.ApiProduct).To(Equal(""))
- })
- })
- })
-
- Context("with usecaching set to false", func() {
- 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"))
- })
- })
- Context("get developerInfo for invalid tenantId and apikey", func() {
- It("should return all right data", func() {
- developerInfo := getDeveloperInfo("wrongid", "wrongapikey")
- Expect(developerInfo.ApiProduct).To(Equal(""))
- })
- })
+var _ = Describe("test getKeyForOrgEnvCache()", func() {
+ It("should return key using org and env", func() {
+ res := getKeyForOrgEnvCache("testorg", "testenv")
+ Expect(res).To(Equal("testorg~testenv"))
})
})
-
-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() {
- a := sql.NullString{String: "null", Valid: false}
- res := getValuesIgnoringNull(a)
- Expect(res).To(Equal(""))
- })
- })
- Context("not null sql value", func() {
- It("should return string", func() {
- a := sql.NullString{String: "sql", Valid: true}
- res := getValuesIgnoringNull(a)
- Expect(res).To(Equal("sql"))
- })
- })
-})
-
-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 6d92ee3..91351ea 100644
--- a/init.go
+++ b/init.go
@@ -20,7 +20,6 @@
"os"
"path/filepath"
"sync"
- "time"
)
const (
@@ -54,10 +53,6 @@
// cache to avoid DB calls for each analytics message
useCaching = "apidanalytics_use_caching"
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
@@ -141,22 +136,6 @@
// 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()
- }
- }()
- }
-
// Create a listener for shutdown event and register callback
h := func(event apid.Event) {
log.Infof("Received ApidShutdown event. %v", event)
@@ -197,9 +176,6 @@
// 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 d0c3a42..9143e5b 100644
--- a/listener.go
+++ b/listener.go
@@ -51,9 +51,11 @@
setDB(db)
if config.GetBool(useCaching) {
- createTenantCache(snapshot)
+ createTenantCache()
log.Debug("Created a local cache" +
" for datasope information")
+ createOrgEnvCache()
+ log.Debug("Created a local cache for org~env Information")
} else {
log.Info("Will not be caching any developer or tenant info " +
"and make a DB call for every analytics msg")
@@ -77,6 +79,10 @@
// map as it has multiple readers
tenantCachelock.Lock()
defer tenantCachelock.Unlock()
+
+ orgEnvCacheLock.Lock()
+ defer orgEnvCacheLock.Unlock()
+
for _, ele := range rows {
var scopeuuid, tenantid string
var org, env string
@@ -86,13 +92,20 @@
ele.Get("env", &env)
if scopeuuid != "" {
tenantCache[scopeuuid] = tenant{
- Org: org,
- Env: env,
- TenantId: tenantid}
+ Org: org,
+ Env: env}
log.Debugf("Refreshed local "+
"tenantCache. Added "+
"scope: "+"%s", scopeuuid)
}
+
+ orgEnv := getKeyForOrgEnvCache(org, env)
+ if orgEnv != "" {
+ orgEnvCache[orgEnv] = true
+ log.Debugf("Refreshed local "+
+ "orgEnvCache. Added "+
+ "orgEnv: "+"%s", orgEnv)
+ }
}
case common.Delete:
rows = append(rows, payload.OldRow)
@@ -100,15 +113,27 @@
// as it has multiple readers
tenantCachelock.Lock()
defer tenantCachelock.Unlock()
+
+ orgEnvCacheLock.Lock()
+ defer orgEnvCacheLock.Unlock()
for _, ele := range rows {
- var scopeuuid string
+ var scopeuuid, org, env string
ele.Get("id", &scopeuuid)
+ ele.Get("org", &org)
+ ele.Get("env", &env)
if scopeuuid != "" {
delete(tenantCache, scopeuuid)
log.Debugf("Refreshed local"+
" tenantCache. Deleted"+
" scope: %s", scopeuuid)
}
+ orgEnv := getKeyForOrgEnvCache(org, env)
+ if orgEnv != "" {
+ delete(orgEnvCache, orgEnv)
+ log.Debugf("Refreshed local"+
+ " orgEnvCache. Deleted"+
+ " org~env: %s", orgEnv)
+ }
}
}
}
diff --git a/listener_test.go b/listener_test.go
index 1de46ba..fc73c2f 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -15,103 +15,61 @@
package apidAnalytics
import (
- . "github.com/onsi/ginkgo"
- . "github.com/onsi/gomega"
-
"github.com/30x/apid-core"
"github.com/apigee-labs/transicator/common"
)
+import (
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+)
+
const (
LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope"
)
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() {
+ var db apid.DB
+ var snapshot common.Snapshot
+
+ BeforeEach(func() {
+ db = getDB()
+ snapshot = common.Snapshot{SnapshotInfo: "test_snapshot"}
+ })
+
+ AfterEach(func() {
+ setDB(db)
+ })
+
It("should set DB to appropriate version", func() {
- config.Set(useCaching, false)
+ handler.Handle(&snapshot)
- event := common.Snapshot{
- SnapshotInfo: "test_snapshot",
- Tables: []common.Table{},
- }
-
- handler.Handle(&event)
-
- expectedDB, err := data.DBVersion(event.SnapshotInfo)
+ expectedDB, err := data.DBVersion(snapshot.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() {
- txn, err := getDB().Begin()
- Expect(err).ShouldNot(HaveOccurred())
- txn.Exec("INSERT INTO edgex_data_scope (id, _change_selector, apid_cluster_id, scope, org, env) "+
- "VALUES"+
- "($1,$2,$3,$4,$5,$6)",
- "i2",
- "c2",
- "a2",
- "s2",
- "o2",
- "e2",
- )
- txn.Commit()
+ BeforeEach(func() {
+ config.Set(useCaching, true)
+ createTenantCache()
+ Expect(len(tenantCache)).To(Equal(1))
+ createOrgEnvCache()
+ Expect(len(orgEnvCache)).To(Equal(1))
+ })
+ AfterEach(func() {
+ config.Set(useCaching, false)
+ })
+
+ It("insert/delete event should add/remove to/from cache if usecaching is true", func() {
insert := common.ChangeList{
LastSequence: "test",
Changes: []common.Change{
@@ -136,14 +94,11 @@
handler.Handle(&insert)
tenant := tenantCache["i2"]
- Expect(tenant.TenantId).To(Equal("s2"))
Expect(tenant.Org).To(Equal("o2"))
Expect(tenant.Env).To(Equal("e2"))
- txn, err = getDB().Begin()
- Expect(err).ShouldNot(HaveOccurred())
- txn.Exec("DELETE FROM edgex_data_scope where id = 'i2'")
- txn.Commit()
+ orgEnv := getKeyForOrgEnvCache("o2", "e2")
+ Expect(orgEnvCache[orgEnv]).To(BeTrue())
delete := common.ChangeList{
LastSequence: "test",
@@ -159,6 +114,9 @@
handler.Handle(&delete)
_, exists := tenantCache["i2"]
Expect(exists).To(BeFalse())
+
+ _, exists = orgEnvCache[orgEnv]
+ Expect(exists).To(BeFalse())
})
})
})
