Merge branch 'master' into all-handlers
diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go index 92c6aff..ef8a1f6 100644 --- a/apigeeSync_suite_test.go +++ b/apigeeSync_suite_test.go
@@ -1,12 +1,389 @@ -package apidApigeeSync_test +package apidApigeeSync import ( . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" "testing" + "github.com/30x/apid" + "github.com/30x/apid/factory" + "io/ioutil" + "net/http/httptest" + "os" + "encoding/json" + "net/http" + "github.com/apigee-labs/transicator/common" + "time" + "strconv" ) +var ( + tmpDir string + testServer *httptest.Server + testRouter apid.Router +) + +const testScope = "bootstrap" + +var _ = BeforeSuite(func(done Done) { + var phase int + + apid.Initialize(factory.DefaultServicesFactory()) + + config := apid.Config() + + var err error + tmpDir, err = ioutil.TempDir("", "api_test") + Expect(err).NotTo(HaveOccurred()) + config.Set("local_storage_path", tmpDir) + + testRouter = apid.API().Router() + testServer = httptest.NewServer(testRouter) + + config.Set(configProxyServerBaseURI, testServer.URL) + config.Set(configSnapServerBaseURI, testServer.URL) + config.Set(configChangeServerBaseURI, testServer.URL) + config.Set(configApidClusterId, "apid_config_scope_0") + config.Set(configName, "testhost") + + config.Set(configSnapshotProtocol, "json") + config.Set(configApidClusterId, testScope) + config.Set(configConsumerKey, "XXXXXXX") + config.Set(configConsumerSecret, "YYYYYYY") + + // mock upstream testServer + testRouter.HandleFunc("/accesstoken", func(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + + Expect(req.Method).To(Equal("POST")) + Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value")) + + err := req.ParseForm() + Expect(err).NotTo(HaveOccurred()) + Expect(req.Form.Get("grant_type")).To(Equal("client_credentials")) + Expect(req.Header.Get("status")).To(Equal("ONLINE")) + Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) + Expect(req.Header.Get("display_name")).To(Equal("testhost")) + + var plugInfo []pluginDetail + plInfo := []byte(req.Header.Get("plugin_details")) + err = json.Unmarshal(plInfo, &plugInfo) + Expect(err).NotTo(HaveOccurred()) + + Expect(plugInfo[0].Name).To(Equal("apidApigeeSync")) + Expect(plugInfo[0].SchemaVersion).To(Equal("0.0.2")) + + res := oauthTokenResp{} + res.AccessToken = "accesstoken" + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) + + }).Methods("POST") + + testRouter.HandleFunc("/snapshots", func(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + + q := req.URL.Query() + + if phase == 0 { + phase = 1 + Expect(q.Get("scope")).To(Equal(testScope)) + Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) + + apidcfgItem := common.Row{} + apidcfgItems := []common.Row{} + apidcfgItemCh := common.Row{} + apidcfgItemsCh := []common.Row{} + scv := &common.ColumnVal{ + Value: testScope, + Type: 1, + } + apidcfgItem["id"] = scv + scv = &common.ColumnVal{ + Value: testScope, + Type: 1, + } + apidcfgItem["_change_selector"] = scv + apidcfgItems = append(apidcfgItems, apidcfgItem) + + scv = &common.ColumnVal{ + Value: "apid_config_scope_id_0", + Type: 1, + } + apidcfgItemCh["id"] = scv + + scv = &common.ColumnVal{ + Value: "apid_config_scope_id_0", + Type: 1, + } + apidcfgItemCh["_change_selector"] = scv + + scv = &common.ColumnVal{ + Value: testScope, + Type: 1, + } + apidcfgItemCh["apid_cluster_id"] = scv + + scv = &common.ColumnVal{ + Value: "ert452", + Type: 1, + } + apidcfgItemCh["scope"] = scv + + { + scv = &common.ColumnVal{ + Value: "att", + Type: 1, + } + apidcfgItemCh["org"] = scv + + } + { + scv = &common.ColumnVal{ + Value: "prod", + Type: 1, + } + apidcfgItemCh["env"] = scv + } + + apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh) + + res := &common.Snapshot{} + res.SnapshotInfo = "snapinfo1" + + res.Tables = []common.Table{ + { + Name: "edgex.apid_cluster", + Rows: apidcfgItems, + }, + { + Name: "edgex.data_scope", + Rows: apidcfgItemsCh, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + + w.Write(body) + return + } else { + phase = 2 + Expect(q.Get("scope")).To(Equal("ert452")) + res := &common.Snapshot{} + res.SnapshotInfo = "snapinfo1" + + apidcfgItems := []common.Row{} + res.Tables = []common.Table{ + { + Name: "kms.api_product", + Rows: apidcfgItems, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + + w.Write(body) + return + } + + }).Methods("GET") + + testRouter.HandleFunc("/changes", func(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + + if req.URL.Query().Get("since") == "lastSeq_01" { + go func() { + block, err := strconv.Atoi(req.URL.Query().Get("block")) + Expect(err).NotTo(HaveOccurred()) + time.Sleep(time.Duration(block) * time.Second) + w.WriteHeader(http.StatusNotModified) + }() + return + } + + Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) + q := req.URL.Query() + Expect(q.Get("snapshot")).To(Equal("snapinfo1")) + scparams := q["scope"] + Expect(scparams).To(ContainElement("ert452")) + Expect(scparams).To(ContainElement("bootstrap")) + + res := &common.ChangeList{} + + res.LastSequence = "lastSeq_01" + mpItems := common.Row{} + + scv := &common.ColumnVal{ + Value: "apid_config_scope_id_1", + Type: 1, + } + mpItems["id"] = scv + + scv = &common.ColumnVal{ + Value: testScope, + Type: 1, + } + mpItems["apid_cluster_id"] = scv + + scv = &common.ColumnVal{ + Value: "ert452", + Type: 1, + } + mpItems["scope"] = scv + { + scv = &common.ColumnVal{ + Value: "att", + Type: 1, + } + mpItems["org"] = scv + } + { + scv = &common.ColumnVal{ + Value: "prod", + Type: 1, + } + mpItems["env"] = scv + } + + res.Changes = []common.Change{ + { + Table: "edgex.data_scope", + NewRow: mpItems, + Operation: 1, + }, + } + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) + + }).Methods("GET") + + // This is actually the first test :) + // Tests that entire bootstrap and set of sync operations work + apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) { + defer GinkgoRecover() + + if s, ok := event.(*common.Snapshot); ok { + + Expect(s.SnapshotInfo).Should(Equal("snapinfo1")) + + for _, t := range s.Tables { + switch t.Name { + + case "edgex.apid_cluster": + Expect(t.Rows).To(HaveLen(1)) + r := t.Rows[0] + var cs, id string + r.Get("_change_selector", &cs) + r.Get("id", &id) + + Expect(cs).To(Equal("bootstrap")) + Expect(id).To(Equal("bootstrap")) + + case "edgex.data_scope": + Expect(t.Rows).To(HaveLen(1)) + r := t.Rows[0] + + var cs, id, clusterID, env, org, scope string + r.Get("_change_selector", &cs) + r.Get("id", &id) + r.Get("apid_cluster_id", &clusterID) + r.Get("env", &env) + r.Get("org", &org) + r.Get("scope", &scope) + + Expect(id).To(Equal("apid_config_scope_id_0")) + Expect(cs).To(Equal("apid_config_scope_id_0")) + Expect(clusterID).To(Equal("bootstrap")) + Expect(env).To(Equal("prod")) + Expect(org).To(Equal("att")) + Expect(scope).To(Equal("ert452")) + + case "kms.api_product": + Expect(t.Rows).To(HaveLen(0)) + + default: + Fail("invalid table: " + t.Name) + } + } + + } else if cl, ok := event.(*common.ChangeList); ok { + + Expect(cl.LastSequence).To(Equal("lastSeq_01")) + Expect(cl.Changes).To(HaveLen(1)) + + c := cl.Changes[0] + Expect(c.Table).To(Equal("edgex.data_scope")) + Expect(c.Operation).To(Equal(common.Insert)) + + Expect(c.NewRow).ToNot(BeNil()) + + var id, clusterID, env, org, scope string + c.NewRow.Get("id", &id) + c.NewRow.Get("apid_cluster_id", &clusterID) + c.NewRow.Get("env", &env) + c.NewRow.Get("org", &org) + c.NewRow.Get("scope", &scope) + + Expect(id).To(Equal("apid_config_scope_id_1")) + Expect(clusterID).To(Equal("bootstrap")) + Expect(env).To(Equal("prod")) + Expect(org).To(Equal("att")) + Expect(scope).To(Equal("ert452")) + + events.ListenFunc(apid.EventDeliveredSelector, func(e apid.Event) { + defer GinkgoRecover() + + // allow other handler to execute to insert last_sequence + time.Sleep(50 * time.Millisecond) + var seq string + err = getDB(). + QueryRow("SELECT last_sequence FROM APID_CLUSTER LIMIT 1;"). + Scan(&seq) + + Expect(err).NotTo(HaveOccurred()) + Expect(seq).To(Equal("lastSeq_01")) + + close(done) + }) + } + }) + + apid.InitializePlugins() +}) + +var _ = BeforeEach(func() { + apid.Events().Close() + + token = "" + downloadDataSnapshot = false + downloadBootSnapshot = false + changeFinished = false + lastSequence = "" + + _, err := getDB().Exec("DELETE FROM APID_CLUSTER") + Expect(err).NotTo(HaveOccurred()) + _, err = getDB().Exec("DELETE FROM DATA_SCOPE") + Expect(err).NotTo(HaveOccurred()) + + db, err := data.DB() + Expect(err).NotTo(HaveOccurred()) + _, err = db.Exec("DELETE FROM APID") + Expect(err).NotTo(HaveOccurred()) +}) + + +var _ = AfterSuite(func() { + apid.Events().Close() + if testServer != nil { + testServer.Close() + } + os.RemoveAll(tmpDir) +}) + func TestApigeeSync(t *testing.T) { RegisterFailHandler(Fail) RunSpecs(t, "ApigeeSync Suite")
diff --git a/apigee_sync.go b/apigee_sync.go index a3de787..be087ec 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -14,43 +14,44 @@ ) var token string -var tokenActive, downloadDataSnapshot, downloadBootSnapshot, chfin bool +var downloadDataSnapshot, downloadBootSnapshot, changeFinished bool var lastSequence string -var gsnapshotInfo string -func add_headers(req *http.Request) { - req.Header.Add("Authorization", "Bearer "+token) - req.Header.Set("apid_instance_id", guuid) - req.Header.Set("apid_cluster_Id", gapidConfigId) +func addHeaders(req *http.Request) { + req.Header.Add("Authorization", "Bearer " + token) + req.Header.Set("apid_instance_id", apidInfo.InstanceID) + req.Header.Set("apid_cluster_Id", apidInfo.ClusterID) req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339)) } -func donehandler(e apid.Event) { - if rsp, ok := e.(apid.EventDeliveryEvent); ok { - if rsp.Description == "event complete" { - if ev, ok := rsp.Event.(*common.Snapshot); ok { - if downloadBootSnapshot == false { - downloadBootSnapshot = true - log.Debug("Updated bootstrap SnapshotInfo") - } else { - gsnapshotInfo = ev.SnapshotInfo - downloadDataSnapshot = true - log.Debug("Updated data SnapshotInfo") - } - } else if ev, ok := rsp.Event.(*common.ChangeList); ok { +func postPluginDataDelivery(e apid.Event) { + + if ede, ok := e.(apid.EventDeliveryEvent); ok { + + if ev, ok := ede.Event.(*common.ChangeList); ok { + if lastSequence != ev.LastSequence { lastSequence = ev.LastSequence - status := persistChange(lastSequence) - if status == false { - log.Fatal("Unable to update Sequence in DB") + err := persistChange(lastSequence) + if err != nil { + log.Panic("Unable to update Sequence in DB") } - chfin = true + } + changeFinished = true + + } else if _, ok := ede.Event.(*common.Snapshot); ok { + if downloadBootSnapshot == false { + downloadBootSnapshot = true + log.Debug("Updated bootstrap SnapshotInfo") + } else { + downloadDataSnapshot = true + log.Debug("Updated data SnapshotInfo") } } } } /* - * Helper function that sleeps for N seconds, if comm. with change agent + * Helper function that sleeps for N seconds if comm with change agent * fails. The retry interval gradually is incremented each time it fails * till it reaches the Polling Int time, and after which it constantly * retries at the polling time interval @@ -61,10 +62,13 @@ pollInterval := config.GetInt(configPollInterval) for { startTime := time.Second - _ = pollChangeAgent() // todo: handle error + err := pollChangeAgent() + if err != nil { + log.Debugf("Error connecting to changeserver: %v", err) + } endTime := time.Second // Gradually increase retry interval, and max at some level - if endTime-startTime <= 1 { + if endTime - startTime <= 1 { if times < pollInterval { times++ } else { @@ -87,7 +91,7 @@ func pollChangeAgent() error { if downloadDataSnapshot != true { - log.Warning("Waiting for snapshot download to complete") + log.Warn("Waiting for snapshot download to complete") return errors.New("Snapshot download in progress...") } changesUri, err := url.Parse(config.GetString(configChangeServerBaseURI)) @@ -101,10 +105,10 @@ * Check to see if we have lastSequence already saved in the DB, * in which case, it has to be used to prevent re-reading same data */ - lastSequence = findapidConfigInfo("lastSequence") + lastSequence = findApidConfigInfo(lastSequence) for { log.Debug("polling...") - if tokenActive == false { + if token == "" { /* token not valid?, get a new token */ status := getBearerToken() if status == false { @@ -113,7 +117,7 @@ } /* Find the scopes associated with the config id */ - scopes := findScopesforId(gapidConfigId) + scopes := findScopesForId(apidInfo.ClusterID) v := url.Values{} /* Sequence added to the query if available */ @@ -130,16 +134,16 @@ for _, scope := range scopes { v.Add("scope", scope) } - v.Add("scope", gapidConfigId) - v.Add("snapshot", gsnapshotInfo) + v.Add("scope", apidInfo.ClusterID) + v.Add("snapshot", apidInfo.LastSnapshot) changesUri.RawQuery = v.Encode() uri := changesUri.String() - log.Info("Fetching changes: ", uri) + log.Debugf("Fetching changes: %s", uri) /* If error, break the loop, and retry after interval */ client := &http.Client{} req, err := http.NewRequest("GET", uri, nil) - add_headers(req) + addHeaders(req) r, err := client.Do(req) if err != nil { log.Errorf("change agent comm error: %s", err) @@ -149,12 +153,11 @@ /* If the call is not Authorized, update flag */ if r.StatusCode != http.StatusOK { if r.StatusCode == http.StatusUnauthorized { - tokenActive = false + token = "" log.Errorf("Token expired? Unauthorized request.") } r.Body.Close() - log.Errorf("Get Changes request failed with Resp err: %d", - r.StatusCode) + log.Errorf("Get Changes request failed with Resp err: %d", r.StatusCode) return err } @@ -162,14 +165,22 @@ err = json.NewDecoder(r.Body).Decode(&resp) r.Body.Close() if err != nil { - log.Errorf("JSON Response Data not parsable: [%s] ", err) + log.Errorf("JSON Response Data not parsable: %v", err) return err } + if lastSequence != resp.LastSequence { + lastSequence = resp.LastSequence + err := persistChange(lastSequence) + if err != nil { + log.Panic("Unable to update Sequence in DB") + } + } + /* If valid data present, Emit to plugins */ if len(resp.Changes) > 0 { - chfin = false - events.ListenFunc(apid.EventDeliveredSelector, donehandler) + changeFinished = false + events.ListenFunc(apid.EventDeliveredSelector, postPluginDataDelivery) events.Emit(ApigeeSyncEventSelector, &resp) /* * The plugins should have finished what they are doing. @@ -178,18 +189,26 @@ * (Should there be a configurable Fudge factor?) FIXME */ for count := 0; count < 1000; count++ { - if chfin == false { - log.Info("Waiting for plugins to complete...") + if changeFinished == false { + log.Debug("Waiting for plugins to complete...") time.Sleep(time.Duration(count) * 100 * time.Millisecond) } else { break } } - if chfin == false { - log.Fatal("Never got ack from plugins. Investigate..") + if changeFinished == false { + log.Panic("Never got ack from plugins. Investigate.") } } else { - log.Info("No Changes detected for Scopes ", scopes) + log.Debugf("No Changes detected for Scopes: %s", scopes) + + if lastSequence != resp.LastSequence { + lastSequence = resp.LastSequence + err := persistChange(lastSequence) + if err != nil { + log.Panic("Unable to update Sequence in DB") + } + } } } } @@ -200,26 +219,27 @@ */ func getBearerToken() bool { - log.Info("Getting a Bearer token.") + log.Debug("Getting a Bearer token.") uri, err := url.Parse(config.GetString(configProxyServerBaseURI)) if err != nil { log.Error(err) return false } uri.Path = path.Join(uri.Path, "/accesstoken") - tokenActive = false + + token = "" form := url.Values{} form.Set("grant_type", "client_credentials") form.Add("client_id", config.GetString(configConsumerKey)) form.Add("client_secret", config.GetString(configConsumerSecret)) req, err := http.NewRequest("POST", uri.String(), bytes.NewBufferString(form.Encode())) req.Header.Set("Content-Type", "application/x-www-form-urlencoded; param=value") - req.Header.Set("display_name", ginstName) - req.Header.Set("apid_instance_id", guuid) - req.Header.Set("apid_cluster_Id", gapidConfigId) + req.Header.Set("display_name", apidInfo.InstanceName) + req.Header.Set("apid_instance_id", apidInfo.InstanceID) + req.Header.Set("apid_cluster_Id", apidInfo.ClusterID) req.Header.Set("status", "ONLINE") req.Header.Set("created_at_apid", time.Now().Format(time.RFC3339)) - req.Header.Set("plugin_details", gpgInfo) + req.Header.Set("plugin_details", apidPluginDetails) client := &http.Client{} resp, err := client.Do(req) @@ -245,8 +265,7 @@ return false } token = oauthResp.AccessToken - tokenActive = true - log.Info("Got a new Bearer token.") + log.Debug("Got a new Bearer token.") return true } @@ -267,7 +286,7 @@ func Redirect(req *http.Request, via []*http.Request) error { req.Header.Add("Authorization", "Bearer "+token) - req.Header.Add("org", gapidConfigId) + req.Header.Add("org", apidInfo.ClusterID) return nil } @@ -283,28 +302,25 @@ * If there is already previous data in sqlite, don't fetch * again from snapshot server. */ -func DownloadSnapshots() { +func bootstrap() { - /* - * Skip Downloading snapshot, if there is already a snapshot - * available from previous run of APID - */ - gsnapshotInfo = findapidConfigInfo("snapshotInfo") - if gsnapshotInfo != "" { + // Skip Downloading snapshot if there is already a snapshot available from previous run of APID + if apidInfo.LastSnapshot != "" { + downloadDataSnapshot = true downloadBootSnapshot = true - log.Infof("Starting on downloaded snapshot: %s", gsnapshotInfo) + log.Infof("Starting on downloaded snapshot: %s", apidInfo.LastSnapshot) - // verify DB is accessible - _, err := data.DBVersion(gsnapshotInfo) + // ensure DB version will be accessible on behalf of dependant plugins + _, err := data.DBVersion(apidInfo.LastSnapshot) if err != nil { log.Panicf("Database inaccessible: %v", err) } - // allow plugins to start immediately on existing database + // allow plugins (including this one) to start immediately on existing database snap := &common.Snapshot{ - SnapshotInfo: gsnapshotInfo, + SnapshotInfo: apidInfo.LastSnapshot, } events.Emit(ApigeeSyncEventSelector, snap) @@ -312,7 +328,7 @@ } /* Phase 1 */ - DownloadSnapshot() + downloadSnapshot() /* * Give some time for all the plugins to process the Downloaded @@ -332,20 +348,22 @@ log.Debug("Proceeding with existing Sqlite data") } else if downloadBootSnapshot == true { log.Debug("Proceed to download Snapshot for data scopes") - DownloadSnapshot() + downloadSnapshot() } else { - log.Fatal("Snapshot for bootscope failed") + log.Panic("Snapshot for bootscope failed") } } -func DownloadSnapshot() { +func downloadSnapshot() { + + log.Debugf("downloadSnapshot") var scopes []string /* Get the bearer token */ status := getBearerToken() if status == false { - log.Fatal("Unable to get Bearer token or is Invalid") + log.Panic("Unable to get Bearer token or is Invalid") } snapshotUri, err := url.Parse(config.GetString(configSnapServerBaseURI)) if err != nil { @@ -353,12 +371,12 @@ } if downloadBootSnapshot == false { - scopes = append(scopes, (gapidConfigId)) + scopes = append(scopes, apidInfo.ClusterID) } else { - scopes = findScopesforId(gapidConfigId) + scopes = findScopesForId(apidInfo.ClusterID) } if scopes == nil { - log.Fatal("Scope cannot be found to download snapshot") + log.Panic("Scope cannot be found to download snapshot") } /* Frame and send the snapshot request */ snapshotUri.Path = path.Join(snapshotUri.Path, "/snapshots") @@ -369,13 +387,13 @@ } snapshotUri.RawQuery = v.Encode() uri := snapshotUri.String() - log.Info("Snapshot Download : ", uri) + log.Info("Snapshot Download: ", uri) client := &http.Client{ CheckRedirect: Redirect, } req, err := http.NewRequest("GET", uri, nil) - add_headers(req) + addHeaders(req) /* Set the transport protocol type based on conf file input */ if config.GetString(configSnapshotProtocol) == "json" { @@ -397,11 +415,10 @@ if err != nil { if downloadBootSnapshot == false { - log.Fatal("JSON Response Data not parsable: ", err) + log.Fatalf("JSON Response Data not parsable: %v", err) } else { - /* - * If the data set is empty, allow it to proceed, as changeserver + * If the data set is empty, allow it to proceed, as change server * will feed data. Since Bootstrapping has passed, it has the * Bootstrap config id to function. */ @@ -412,7 +429,7 @@ if r.StatusCode == 200 { log.Info("Emit Snapshot response to plugins") - events.ListenFunc(apid.EventDeliveredSelector, donehandler) + events.ListenFunc(apid.EventDeliveredSelector, postPluginDataDelivery) events.Emit(ApigeeSyncEventSelector, &resp) } else { @@ -420,88 +437,3 @@ } } - -/* - * For the given apidConfigId, this function will retrieve all the scopes - * associated with it - */ -func findScopesforId(configId string) (scopes []string) { - - var scope string - db, err := data.DB() - if err != nil { - log.Errorf("DB open Error: %s", err) - return nil - } - - rows, err := db.Query("select scope from DATA_SCOPE where apid_cluster_id = $1", configId) - if err != nil { - log.Errorf("Failed to query DATA_SCOPE. Err: %s", err) - return nil - } - defer rows.Close() - for rows.Next() { - rows.Scan(&scope) - scopes = append(scopes, scope) - } - return scopes -} - -/* - * Retrieve SnapshotInfo for the given apidConfigId from apid_config table - */ -func findapidConfigInfo(qparam string) (info string) { - - db, err := data.DB() - if err != nil { - log.Errorf("DB open Error: %s", err) - return "" - } - query := "select " + qparam + " from APID_CLUSTER" - rows, err := db.Query(query) - if err != nil { - log.Errorf("Failed to query APID_CLUSTER. Err: %s", err) - return "" - } - defer rows.Close() - for rows.Next() { - rows.Scan(&info) - } - return info -} - -/* - * Persist the last change Id each time a change has been successfully - * processed by the plugin(s) - */ -func persistChange(lastChange string) bool { - db, err := data.DB() - if err != nil { - log.Errorf("DB open Error: %s", err) - return false - } - txn, err := db.Begin() - if err != nil { - log.Error("Unable to create Sqlite transaction") - return false - } - prep, err := txn.Prepare("UPDATE APID_CLUSTER SET lastSequence=$1;") - if err != nil { - log.Error("UPDATE APID_CLUSTER Failed: ", err) - return false - } - defer prep.Close() - s := txn.Stmt(prep) - _, err = s.Exec(lastChange) - s.Close() - if err != nil { - log.Error("UPDATE DATA_SCOPE Failed: ", err) - txn.Rollback() - return false - } else { - log.Info("UPDATE DATA_SCOPE Success: (", lastChange, ")") - txn.Commit() - return true - } - -}
diff --git a/apigee_sync_test.go b/apigee_sync_test.go index 507aa99..dea238c 100644 --- a/apigee_sync_test.go +++ b/apigee_sync_test.go
@@ -1,314 +1,29 @@ package apidApigeeSync import ( - "encoding/json" - "github.com/30x/apid" - "github.com/30x/apid/factory" - "github.com/apigee-labs/transicator/common" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" - "io/ioutil" - "net/http" - "net/http/httptest" - "os" - "time" + "github.com/30x/apid" + "github.com/apigee-labs/transicator/common" ) -var _ = Describe("api", func() { +var _ = Describe("listener", func() { - var server *httptest.Server - var plugInfo []pluginDetail + It("should bootstrap from local DB if present", func(done Done) { - BeforeSuite(func() { - apid.Initialize(factory.DefaultServicesFactory()) - }) + Expect(apidInfo.LastSnapshot).NotTo(BeEmpty()) - AfterSuite(func() { - apid.Events().Close() - server.Close() - }) + apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) { + defer GinkgoRecover() - It("perform sync round-trip", func(done Done) { - scount := 0 - phase := 0 - scope := "bootstrap" - key := "XXXXXXX" - secret := "YYYYYYY" + if s, ok := event.(*common.Snapshot); ok { + Expect(s.SnapshotInfo).Should(Equal(apidInfo.LastSnapshot)) + Expect(s.Tables).To(BeNil()) - // mock upstream server - server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { - - // first request is for a token - if req.URL.Path == "/accesstoken" { - Expect(req.Method).To(Equal("POST")) - Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value")) - - err := req.ParseForm() - Expect(err).NotTo(HaveOccurred()) - Expect(req.Form.Get("grant_type")).To(Equal("client_credentials")) - Expect(req.Header.Get("status")).To(Equal("ONLINE")) - Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) - Expect(req.Header.Get("display_name")).To(Equal("testhost")) - plinfo := []byte(req.Header.Get("plugin_details")) - err = json.Unmarshal(plinfo, &plugInfo) - Expect(err).NotTo(HaveOccurred()) - Expect(plugInfo[0].Name).To(Equal("apidApigeeSync")) - Expect(plugInfo[0].SchemaVersion).To(Equal("0.0.2")) - - res := oauthTokenResp{} - res.AccessToken = "accesstoken" - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - w.Write(body) - return + close(done) } + }) - // next requests are for changes - if req.URL.Path == "/snapshots" { - Expect(req.Method).To(Equal("GET")) - q := req.URL.Query() - - if phase == 0 { - phase = 1 - Expect(q.Get("scope")).To(Equal(scope)) - Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) - - apidcfgItem := common.Row{} - apidcfgItems := []common.Row{} - apidcfgItemCh := common.Row{} - apidcfgItemsCh := []common.Row{} - scv := &common.ColumnVal{ - Value: scope, - Type: 1, - } - apidcfgItem["id"] = scv - scv = &common.ColumnVal{ - Value: scope, - Type: 1, - } - apidcfgItem["_change_selector"] = scv - apidcfgItems = append(apidcfgItems, apidcfgItem) - - scv = &common.ColumnVal{ - Value: "apid_config_scope_id_0", - Type: 1, - } - apidcfgItemCh["id"] = scv - - scv = &common.ColumnVal{ - Value: "apid_config_scope_id_0", - Type: 1, - } - apidcfgItemCh["_change_selector"] = scv - - scv = &common.ColumnVal{ - Value: scope, - Type: 1, - } - apidcfgItemCh["apid_cluster_id"] = scv - - scv = &common.ColumnVal{ - Value: "ert452", - Type: 1, - } - apidcfgItemCh["scope"] = scv - - { - scv = &common.ColumnVal{ - Value: "att", - Type: 1, - } - apidcfgItemCh["org"] = scv - - } - { - scv = &common.ColumnVal{ - Value: "prod", - Type: 1, - } - apidcfgItemCh["env"] = scv - } - - apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh) - - res := &common.Snapshot{} - res.SnapshotInfo = "snapinfo1" - - res.Tables = []common.Table{ - { - Name: "edgex.apid_cluster", - Rows: apidcfgItems, - }, - { - Name: "edgex.data_scope", - Rows: apidcfgItemsCh, - }, - } - - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - w.Write(body) - return - } else { - phase = 2 - Expect(q.Get("scope")).To(Equal("ert452")) - res := &common.Snapshot{} - res.SnapshotInfo = "snapinfo1" - - apidcfgItems := []common.Row{} - res.Tables = []common.Table{ - { - Name: "kms.api_product", - Rows: apidcfgItems, - }, - } - - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - w.Write(body) - return - } - - } - // next requests are for changes - if req.URL.Path == "/changes" { - Expect(req.Method).To(Equal("GET")) - Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) - q := req.URL.Query() - Expect(q.Get("snapshot")).To(Equal("snapinfo1")) - scparams := q["scope"] - Expect(scparams).To(ContainElement("ert452")) - Expect(scparams).To(ContainElement("bootstrap")) - - res := &common.ChangeList{} - - res.LastSequence = "lastSeq_01" - mpItems := common.Row{} - - scv := &common.ColumnVal{ - Value: "apid_config_scope_id_1", - Type: 1, - } - mpItems["id"] = scv - - scv = &common.ColumnVal{ - Value: scope, - Type: 1, - } - mpItems["apid_cluster_id"] = scv - - scv = &common.ColumnVal{ - Value: "ert452", - Type: 1, - } - mpItems["scope"] = scv - { - scv = &common.ColumnVal{ - Value: "att", - Type: 1, - } - mpItems["org"] = scv - } - { - scv = &common.ColumnVal{ - Value: "prod", - Type: 1, - } - mpItems["env"] = scv - } - - res.Changes = []common.Change{ - { - Table: "edgex.data_scope", - NewRow: mpItems, - Operation: 1, - }, - } - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - w.Write(body) - return - } - Fail("should not reach") - })) - - config = apid.Config() - config.Set(configProxyServerBaseURI, server.URL) - config.Set(configSnapServerBaseURI, server.URL) - config.Set(configChangeServerBaseURI, server.URL) - config.Set(configApidClusterId, "apid_config_scope_0") - config.Set(configName, "testhost") - - config.Set(configSnapshotProtocol, "json") - config.Set(configApidClusterId, scope) - config.Set(configConsumerKey, key) - config.Set(configConsumerSecret, secret) - - // set up temporary test database - tmpDir, err := ioutil.TempDir("", "apigee_sync_test") - Expect(err).NotTo(HaveOccurred()) - defer os.RemoveAll(tmpDir) - - config.Set("data_path", tmpDir) - - // start process - plugin will automatically start polling - apid.InitializePlugins() - - h := &test_handler{ - "sync data", - func(event apid.Event) { - _, ok := event.(*common.Snapshot) - if ok { - if phase > 1 { - db, err := data.DB() - Expect(err).NotTo(HaveOccurred()) - // verify event data (post snapshot) - err = db.QueryRow("Select count(scp.id) from data_scope as scp INNER JOIN apid_cluster as ap WHERE scp.apid_cluster_id = ap.id").Scan(&scount) - Expect(err).NotTo(HaveOccurred()) - Expect(scount).Should(Equal(1)) - } - } else { - // verify event data (post change) - // There should be 2 scopes now - _, ok := event.(*common.ChangeList) - if ok { - time.Sleep(200 * time.Millisecond) - db, err := data.DB() - Expect(err).NotTo(HaveOccurred()) - err = db.QueryRow("Select count(scp.id) from data_scope as scp INNER JOIN apid_cluster as ap WHERE scp.apid_cluster_id = ap.id").Scan(&scount) - Expect(err).NotTo(HaveOccurred()) - Expect(scount).Should(Equal(2)) - close(done) - } else { - Fail("Unexpected event") - } - } - - }, - } - - donehandler := func(e apid.Event) { - if rsp, ok := e.(apid.EventDeliveryEvent); ok { - Expect(rsp.Description).Should(Equal("event complete")) - } else { - Fail("Unexpected event") - } - } - apid.Events().Listen(ApigeeSyncEventSelector, h) - events.ListenFunc(apid.EventDeliveredSelector, donehandler) - + bootstrap() }) }) - -type test_handler struct { - description string - f func(event apid.Event) -} - -func (t *test_handler) String() string { - return t.description -} - -func (t *test_handler) Handle(event apid.Event) { - t.f(event) -}
diff --git a/data.go b/data.go new file mode 100644 index 0000000..e93971b --- /dev/null +++ b/data.go
@@ -0,0 +1,318 @@ +package apidApigeeSync + +import ( + "database/sql" + "github.com/30x/apid" + "sync" + "fmt" + "crypto/rand" + "errors" +) + +var ( + unsafeDB apid.DB + dbMux sync.RWMutex +) + +type dataApidCluster struct { + ChangeSelector, ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string + Updated, Created string +} + +type dataDataScope struct { + ChangeSelector, ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string + Updated, Created string +} + +/* +This plugin uses 2 databases: +1. The default DB is used for APID table. +2. The versioned DB is used for APID_CLUSTER & DATA_SCOPE +(Currently, the snapshot never changes, but this is future-proof) +*/ +func initDB(db apid.DB) error { + _, err := db.Exec(` + CREATE TABLE IF NOT EXISTS APID ( + instance_id text, + last_snapshot_info text, + PRIMARY KEY (instance_id) + ); + CREATE TABLE IF NOT EXISTS APID_CLUSTER ( + id text, + name text, + description text, + umbrella_org_app_name text, + created text, + created_by text, + updated text, + updated_by text, + _change_selector text, + last_sequence text, + PRIMARY KEY (id) + ); + CREATE TABLE IF NOT EXISTS DATA_SCOPE ( + id text, + apid_cluster_id text, + scope text, + org text, + env text, + created text, + created_by text, + updated text, + updated_by text, + _change_selector text, + PRIMARY KEY (id, apid_cluster_id) + ); + `) + if err != nil { + return err + } + + log.Debug("Database tables created.") + return nil +} + +func getDB() apid.DB { + dbMux.RLock() + db := unsafeDB + dbMux.RUnlock() + return db +} + +func setDB(db apid.DB) { + dbMux.Lock() + unsafeDB = db + dbMux.Unlock() +} + +func insertApidCluster(dac dataApidCluster, txn *sql.Tx) error { + + log.Debugf("inserting into APID_CLUSTER: %v", dac) + + stmt, err := txn.Prepare(` + INSERT INTO APID_CLUSTER + (id, _change_selector, name, umbrella_org_app_name, + created, created_by, updated, updated_by, + description) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9); + `) + if err != nil { + log.Errorf("prepare insert into APID_CLUSTER transaction Failed: %v", err) + return err + } + defer stmt.Close() + + _, err = stmt.Exec( + dac.ID, dac.ChangeSelector, dac.Name, dac.OrgAppName, + dac.Created, dac.CreatedBy, dac.Updated, dac.UpdatedBy, + dac.Description) + + if err != nil { + log.Errorf("insert APID_CLUSTER failed: %v", err) + } + + return err +} + +func insertDataScope(ds dataDataScope, txn *sql.Tx) error { + + log.Debugf("insert DATA_SCOPE: %v", ds) + + stmt, err := txn.Prepare(` + INSERT INTO DATA_SCOPE + (id, apid_cluster_id, scope, org, + env, created, created_by, updated, + updated_by, _change_selector) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10); + `) + if err != nil { + log.Errorf("insert DATA_SCOPE failed: %v", err) + return err + } + defer stmt.Close() + + _, err = stmt.Exec( + ds.ID, ds.ClusterID, ds.Scope, ds.Org, + ds.Env, ds.Created, ds.CreatedBy, ds.Updated, + ds.UpdatedBy, ds.ChangeSelector) + + if err != nil { + log.Errorf("insert DATA_SCOPE failed: %v", err) + return err + } + + return nil +} + +func deleteDataScope(ds dataDataScope, txn *sql.Tx) error { + + log.Debugf("delete DATA_SCOPE: %v", ds) + + stmt, err := txn.Prepare("DELETE FROM DATA_SCOPE WHERE id=$1 and apid_cluster_id=$2") + if err != nil { + log.Errorf("update DATA_SCOPE failed: %v", err) + return err + } + defer stmt.Close() + + _, err = stmt.Exec(ds.ID, ds.ClusterID) + + if err != nil { + log.Errorf("delete DATA_SCOPE failed: %v", err) + return err + } + + return nil +} + +/* + * For the given apidConfigId, this function will retrieve all the scopes + * associated with it + */ +func findScopesForId(configId string) (scopes []string) { + + log.Debugf("findScopesForId: %s", configId) + + var scope string + db := getDB() + + rows, err := db.Query("select scope from DATA_SCOPE where apid_cluster_id = $1", configId) + if err != nil { + log.Errorf("Failed to query DATA_SCOPE: %v", err) + return + } + defer rows.Close() + for rows.Next() { + rows.Scan(&scope) + scopes = append(scopes, scope) + } + + log.Debugf("scopes: %v", scopes) + return +} + +/* + * Retrieve SnapshotInfo for the given apidConfigId from apid_config table + */ +func findApidConfigInfo(qparam string) (info string) { + + log.Debugf("findApidConfigInfo: %s", qparam) + + db := getDB() + + rows, err := db.Query("select ? from APID_CLUSTER", qparam) + if err != nil { + log.Errorf("Failed to query APID_CLUSTER: %v", err) + return + } + defer rows.Close() + for rows.Next() { + rows.Scan(&info) + } + + log.Debugf("info: %s", info) + + return +} + +/* + * Persist the last change Id each time a change has been successfully + * processed by the plugin(s) + */ +func persistChange(lastChange string) error { + + log.Debugf("persistChange: %s", lastChange) + + db := getDB() + + stmt, err := db.Prepare("UPDATE APID_CLUSTER SET last_sequence=$1;") + if err != nil { + log.Errorf("UPDATE APID_CLUSTER Failed: %v", err) + return err + } + defer stmt.Close() + + _, err = stmt.Exec(lastChange) + if err != nil { + log.Errorf("UPDATE DATA_SCOPE Failed: %v", err) + return err + } + + log.Infof("UPDATE DATA_SCOPE Success: %s", lastChange) + + return nil +} + +func getApidInstanceInfo() (info apidInstanceInfo, err error) { + + // always use default database for this + var db apid.DB + db, err = data.DB() + if err != nil { + return + } + + err = db.QueryRow("SELECT instance_id, last_snapshot_info FROM APID LIMIT 1"). + Scan(&info.InstanceID, &info.LastSnapshot) + if err != nil { + if err != sql.ErrNoRows { + log.Errorf("Unable to retrieve apidInstanceInfo: %v", err) + return + } else { + // first start - no row, generate a UUID and store it + err = nil + info.InstanceID = generateUUID() + + db.Exec("INSERT INTO APID (instance_id) VALUES (?)", info.InstanceID) + } + } + + // if name not explicitly configured, just use InstanceID + config.SetDefault(configName, info.InstanceID) + info.InstanceName = config.GetString(configName) + + // not stored in DB + info.ClusterID = config.GetString(configApidClusterId) + + return +} + +func updateApidInstanceInfo() error { + + // always use default database for this + db, err := data.DB() + if err != nil { + return err + } + + rows, err := db.Exec(` + INSERT OR REPLACE + INTO APID (instance_id, last_snapshot_info) + VALUES (?, ?)`, + apidInfo.InstanceID, apidInfo.LastSnapshot) + if err != nil { + return err + } + n, err := rows.RowsAffected() + if err == nil && n == 0 { + err = errors.New("no rows affected") + } + + return err +} + +/* + * generates a random uuid (mix of timestamp & crypto random string) + */ +func generateUUID() string { + + buff := make([]byte, 16) + numRead, err := rand.Read(buff) + if numRead != len(buff) || err != nil { + panic(err) + } + /* uuid v4 spec */ + buff[6] = (buff[6] | 0x40) & 0x4F + buff[8] = (buff[8] | 0x80) & 0xBF + return fmt.Sprintf("%x-%x-%x-%x-%x", buff[0:4], buff[4:6], buff[6:8], buff[8:10], buff[10:]) +}
diff --git a/init.go b/init.go index d143dde..931a9ec 100644 --- a/init.go +++ b/init.go
@@ -1,7 +1,6 @@ package apidApigeeSync import ( - "crypto/rand" "encoding/json" "fmt" "github.com/30x/apid" @@ -21,47 +20,81 @@ ) var ( - log apid.LogService - config apid.ConfigService - data apid.DataService - events apid.EventsService - gapidConfigId string - guuid string - ginstName string - gpgInfo string + log apid.LogService + config apid.ConfigService + data apid.DataService + events apid.EventsService + + apidInfo apidInstanceInfo + apidPluginDetails string ) +type apidInstanceInfo struct { + InstanceID, InstanceName, ClusterID, LastSnapshot string +} + type pluginDetail struct { Name string `json:"name"` SchemaVersion string `json:"schemaVer"` } -/* - * generates a random uuid - */ -func generate_uuid() string { - buff := make([]byte, 16) - numRead, err := rand.Read(buff) - if numRead != len(buff) || err != nil { - panic(err) - } - /* uuid v4 spec */ - buff[6] = (buff[6] | 0x40) & 0x4F - buff[8] = (buff[8] | 0x80) & 0xBF - return fmt.Sprintf("%x-%x-%x-%x-%x", buff[0:4], buff[4:6], buff[6:8], buff[8:10], buff[10:]) -} - func init() { apid.RegisterPlugin(initPlugin) } +func initPlugin(services apid.Services) (apid.PluginData, error) { + log = services.Log().ForModule("apigeeSync") + log.Debug("start init") + + config = services.Config() + config.SetDefault(configPollInterval, 120) + + data = services.Data() + events = services.Events() + + /* This callback function will get called, once all the plugins are + * initialized (not just this plugin). This is needed because, + * downloadSnapshots/changes etc have to begin to be processed only + * after all the plugins are initialized + */ + events.ListenFunc(apid.SystemEventsSelector, postInitPlugins) + + // check for required values + for _, key := range []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret, + configSnapServerBaseURI, configChangeServerBaseURI} { + if !config.IsSet(key) { + return pluginData, fmt.Errorf("Missing required config value: %s", key) + } + } + + // set up default database + db, err := data.DB() + if err != nil { + log.Panicf("Unable to access DB: %v", err) + } + err = initDB(db) + if err != nil { + log.Panicf("Unable to initialize DB: %v", err) + } + setDB(db) + + apidInfo, err = getApidInstanceInfo() + if err != nil { + log.Panicf("Unable to get apid instance info: %v", err) + } + + log.Debug("end init") + + return pluginData, nil +} + +// Plugins have all initialized, gather their info and start the ApigeeSync downloads func postInitPlugins(event apid.Event) { var plinfoDetails []pluginDetail if pie, ok := event.(apid.PluginsInitializedEvent); ok { - /* * Store the plugin details in the heap. Needed during - * Bearer token generation request + * Bearer token generation request. */ for _, plugin := range pie.Plugins { name := plugin.Name @@ -76,17 +109,17 @@ } if plinfoDetails == nil { log.Panicf("No Plugins registered!") - } else { - pgInfo, err := json.Marshal(plinfoDetails) - if err != nil { - log.Panic("Unable to masrhal plugin data", err) - } - gpgInfo = (string(pgInfo[:])) } + pgInfo, err := json.Marshal(plinfoDetails) + if err != nil { + log.Panicf("Unable to marshal plugin data: %v", err) + } + apidPluginDetails = string(pgInfo[:]) + log.Debug("start post plugin init") - /* call to Download Snapshot info */ - go DownloadSnapshots() + + go bootstrap() /* Begin Looking for changes periodically */ log.Debug("starting update goroutine") @@ -97,91 +130,3 @@ } } -func initPlugin(services apid.Services) (apid.PluginData, error) { - log = services.Log().ForModule("apigeeSync") - log.Debug("start init") - - config = services.Config() - data = services.Data() - events = services.Events() - guuid = findapidConfigInfo("instance_id") - if guuid == "" { - guuid = generate_uuid() - } - - /* If The Instance has no name configured, just re-use UUID */ - ginstName = config.GetString(configName) - if ginstName == "" { - ginstName = guuid - } - - /* This callback function will get called, once all the plugins are - * initialized (not just this plugin). This is needed because, - * DownloadSnapshots/Changes etc have to begin to be processed only - * after all the plugins are initialized - */ - events.ListenFunc(apid.SystemEventsSelector, postInitPlugins) - - config.SetDefault(configPollInterval, 120) - gapidConfigId = config.GetString(configApidClusterId) - db, err := data.DB() - if err != nil { - log.Panic("Unable to access DB", err) - } - - // check for required values - for _, key := range []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret, configSnapServerBaseURI, configChangeServerBaseURI} { - if !config.IsSet(key) { - return pluginData, fmt.Errorf("Missing required config value: %s", key) - } - } - - var count int - row := db.QueryRow("SELECT count(*) FROM sqlite_master WHERE type='table' AND name='apid_cluster' COLLATE NOCASE;") - if err := row.Scan(&count); err != nil { - log.Panic("Unable to setup database", err) - } - if count == 0 { - createTables(db) - } - - log.Debug("end init") - - return pluginData, nil -} - -func createTables(db apid.DB) { - _, err := db.Exec(` -CREATE TABLE apid_cluster ( - id text, - instance_id text, - name text, - description text, - umbrella_org_app_name text, - created int64, - created_by text, - updated int64, - updated_by text, - _change_selector text, - snapshotInfo text, - lastSequence text, - PRIMARY KEY (id) -); -CREATE TABLE data_scope ( - id text, - apid_cluster_id text, - scope text, - org text, - env text, - created int64, - created_by text, - updated int64, - updated_by text, - _change_selector text, - PRIMARY KEY (id) -); -`) - if err != nil { - log.Panic("Unable to initialize DB", err) - } -}
diff --git a/listener.go b/listener.go index c5f154a..41b7a8e 100644 --- a/listener.go +++ b/listener.go
@@ -1,11 +1,15 @@ package apidApigeeSync import ( - "database/sql" "github.com/30x/apid" "github.com/apigee-labs/transicator/common" ) +const ( + LISTENER_TABLE_APID_CLUSTER = "edgex.apid_cluster" + LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope" +) + type handler struct { } @@ -15,180 +19,150 @@ func (h *handler) Handle(e apid.Event) { - res := true - - db, err := data.DB() - if err != nil { - panic("Unable to access Sqlite DB") - } - - txn, err := db.Begin() - if err != nil { - log.Error("Unable to create Sqlite transaction") - return - } - - snapData, ok := e.(*common.Snapshot) - if ok { - res = processSnapshot(snapData, txn) + if changeSet, ok := e.(*common.ChangeList); ok { + processChangeList(changeSet) + } else if snapShot, ok := e.(*common.Snapshot); ok { + processSnapshot(snapShot) } else { - changeSet, ok := e.(*common.ChangeList) - if ok { - res = processChange(changeSet, txn) - } else { - log.Fatal("Received invalid event: %v", e) - } + log.Errorf("Received invalid event. Ignoring. %v", e) } - if res == true { - txn.Commit() - } else { - txn.Rollback() - } - return } -func processSnapshot(snapshot *common.Snapshot, txn *sql.Tx) bool { +func processSnapshot(snapshot *common.Snapshot) { - log.Debugf("Process Snapshot data") - res := true + log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo) - for _, payload := range snapshot.Tables { - - switch payload.Name { - case "edgex.apid_cluster": - res = insertApidCluster(payload.Rows, txn, snapshot.SnapshotInfo) - case "edgex.data_scope": - res = insertDataScopes(payload.Rows, txn) - } - if res == false { - log.Error("Error encountered in Downloading Snapshot for ApidApigeeSync") - return res - } + db, err := data.DBVersion(snapshot.SnapshotInfo) + if err != nil { + log.Panicf("Unable to access database: %v", err) } - return res -} -func processChange(changes *common.ChangeList, txn *sql.Tx) bool { + err = initDB(db) + if err != nil { + log.Panicf("Unable to initialize database: %v", err) + } - log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes)) - var rows []common.Row - res := true + tx, err := db.Begin() + if err != nil { + log.Panicf("Error starting transaction: %v", err) + } + defer tx.Rollback() - for _, payload := range changes.Changes { - rows = nil - switch payload.Table { - case "edgex.data_scope": - switch payload.Operation { - case common.Insert: - rows = append(rows, payload.NewRow) - res = insertDataScopes(rows, txn) + for _, table := range snapshot.Tables { + + switch table.Name { + case LISTENER_TABLE_APID_CLUSTER: + if len(table.Rows) != 1 { + log.Panic("Illegal state for apid_cluster. Must be a single row.") + } + ac := makeApidClusterFromRow(table.Rows[0]) + err := insertApidCluster(ac, tx) + if err != nil { + log.Panic("Snapshot update failed: %v", err) + } + + case LISTENER_TABLE_DATA_SCOPE: + for _, row := range table.Rows { + ds := makeDataScopeFromRow(row) + err := insertDataScope(ds, tx) + if err != nil { + log.Panic("Snapshot update failed: %v", err) + } } } - if res == false { - log.Error("Sql Operation error. Operation rollbacked") - return res - } } - return res -} -/* - * INSERT INTO APP_CREDENTIAL op - */ -func insertApidCluster(rows []common.Row, txn *sql.Tx, snapInfo string) bool { - - var scope, id, name, orgAppName, createdBy, updatedBy, Description string - var updated, created int64 - - prep, err := txn.Prepare("INSERT INTO APID_CLUSTER (id, instance_id, _change_selector, name, umbrella_org_app_name, created, created_by, updated, updated_by, snapshotInfo)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10);") + err = tx.Commit() if err != nil { - log.Error("INSERT APID_CLUSTER Failed: ", err) - return false + log.Panicf("Error committing Snapshot change: %v", err) } - defer prep.Close() - for _, ele := range rows { - ele.Get("id", &id) - ele.Get("_change_selector", &scope) - ele.Get("name", &name) - ele.Get("umbrella_org_app_name", &orgAppName) - ele.Get("created", &created) - ele.Get("created_by", &createdBy) - ele.Get("updated", &updated) - ele.Get("updated_by", &updatedBy) - ele.Get("description", &Description) - - s := txn.Stmt(prep) - _, err = s.Exec( - id, - guuid, - scope, - name, - orgAppName, - created, - createdBy, - updated, - updatedBy, - snapInfo) - s.Close() - if err != nil { - log.Error("INSERT APID_CLUSTER Failed: ", id, ", ", scope, ")", err) - return false - } else { - log.Info("INSERT APID_CLUSTER Success: (", id, ", ", scope, ")") - } - } - return true -} - -/* - * INSERT INTO APP_CREDENTIAL op - */ -func insertDataScopes(rows []common.Row, txn *sql.Tx) bool { - - var id, scopeId, apiConfigId, scope, createdBy, updatedBy, org, env string - var created, updated int64 - - prep, err := txn.Prepare("INSERT INTO DATA_SCOPE (id, _change_selector, apid_cluster_id, scope, org, env, created, created_by, updated, updated_by)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10);") + apidInfo.LastSnapshot = snapshot.SnapshotInfo + err = updateApidInstanceInfo() if err != nil { - log.Error("INSERT DATA_SCOPE Failed: ", err) - return false + log.Panicf("Unable to update instance info: %v", err) } - defer prep.Close() - for _, ele := range rows { + setDB(db) + log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo) +} - ele.Get("id", &id) - ele.Get("_change_selector", &scopeId) - ele.Get("apid_cluster_id", &apiConfigId) - ele.Get("scope", &scope) - ele.Get("org", &org) - ele.Get("env", &env) - ele.Get("created", &created) - ele.Get("created_by", &createdBy) - ele.Get("updated", &updated) - ele.Get("updated_by", &updatedBy) +func processChangeList(changes *common.ChangeList) { - s := txn.Stmt(prep) - _, err = s.Exec( - id, - scopeId, - apiConfigId, - scope, - org, - env, - created, - createdBy, - updated, - updatedBy) - s.Close() + tx, err := getDB().Begin() + if err != nil { + log.Panicf("Error processing ChangeList: %v", err) + } + defer tx.Rollback() - if err != nil { - log.Error("INSERT DATA_SCOPE Failed: ", id, ", ", scope, ")", err) - return false - } else { - log.Info("INSERT DATA_SCOPE Success: (", id, ", ", scope, ")") + log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes)) + + for _, change := range changes.Changes { + switch change.Table { + case "edgex.apid_cluster": + switch change.Operation { + case common.Delete: + // todo: shut down apid, delete databases, scorch the earth! + log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) + default: + log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) + } + case "edgex.data_scope": + switch change.Operation { + case common.Insert: + ds := makeDataScopeFromRow(change.NewRow) + err = insertDataScope(ds, tx) + case common.Delete: + ds := makeDataScopeFromRow(change.OldRow) + deleteDataScope(ds, tx) + default: + // common.Update is not allowed + log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) + } + } + if err != nil{ + log.Panicf("Error processing ChangeList: %v", err) } } - return true + + err = tx.Commit() + if err != nil { + log.Panicf("Error processing ChangeList: %v", err) + } } + +func makeApidClusterFromRow(row common.Row) dataApidCluster { + + dac := dataApidCluster{} + + row.Get("id", &dac.ID) + row.Get("_change_selector", &dac.ChangeSelector) + row.Get("name", &dac.Name) + row.Get("umbrella_org_app_name", &dac.OrgAppName) + row.Get("created", &dac.Created) + row.Get("created_by", &dac.CreatedBy) + row.Get("updated", &dac.Updated) + row.Get("updated_by", &dac.UpdatedBy) + row.Get("description", &dac.Description) + + return dac +} + +func makeDataScopeFromRow(row common.Row) dataDataScope { + + ds := dataDataScope{} + + row.Get("id", &ds.ID) + row.Get("_change_selector", &ds.ChangeSelector) + row.Get("apid_cluster_id", &ds.ClusterID) + row.Get("scope", &ds.Scope) + row.Get("org", &ds.Org) + row.Get("env", &ds.Env) + row.Get("created", &ds.Created) + row.Get("created_by", &ds.CreatedBy) + row.Get("updated", &ds.Updated) + row.Get("updated_by", &ds.UpdatedBy) + + return ds +} +
diff --git a/listener_test.go b/listener_test.go new file mode 100644 index 0000000..02d255d --- /dev/null +++ b/listener_test.go
@@ -0,0 +1,338 @@ +package apidApigeeSync + +import ( + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + + "github.com/apigee-labs/transicator/common" +) + +var _ = Describe("listener", func() { + + handler := handler{} + + Context("ApigeeSync snapshot event", func() { + + It("should set DB to appropriate version", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot", + Tables: []common.Table{}, + } + + handler.Handle(&event) + + Expect(apidInfo.LastSnapshot).To(Equal(event.SnapshotInfo)) + + expectedDB, err := data.DBVersion(event.SnapshotInfo) + Expect(err).NotTo(HaveOccurred()) + + Expect(getDB() == expectedDB).Should(BeTrue()) + }) + + It("should fail if zero apid_cluster rows", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot_fail", + Tables: []common.Table{ + { + Name: LISTENER_TABLE_APID_CLUSTER, + Rows: []common.Row{}, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + It("should fail if more than one apid_cluster rows", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot_fail", + Tables: []common.Table{ + { + Name: LISTENER_TABLE_APID_CLUSTER, + Rows: []common.Row{{}, {}}, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + It("should process a valid Snapshot", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot_valid", + Tables: []common.Table{ + { + Name: LISTENER_TABLE_APID_CLUSTER, + Rows: []common.Row{ + { + "id": &common.ColumnVal{Value: "i"}, + "_change_selector": &common.ColumnVal{Value: "c"}, + "name": &common.ColumnVal{Value: "n"}, + "umbrella_org_app_name": &common.ColumnVal{Value: "o"}, + "created": &common.ColumnVal{Value: "c"}, + "created_by": &common.ColumnVal{Value: "c"}, + "updated": &common.ColumnVal{Value: "u"}, + "updated_by": &common.ColumnVal{Value: "u"}, + "description": &common.ColumnVal{Value: "d"}, + }, + }, + }, + { + 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) + + info, err := getApidInstanceInfo() + Expect(err).NotTo(HaveOccurred()) + + Expect(info.LastSnapshot).To(Equal(event.SnapshotInfo)) + + db := getDB() + + // apid Cluster + var dcs []dataApidCluster + + rows, err := db.Query(` + SELECT id, name, description, umbrella_org_app_name, + created, created_by, updated, updated_by, + _change_selector + FROM APID_CLUSTER`) + Expect(err).NotTo(HaveOccurred()) + defer rows.Close() + + c := dataApidCluster{} + for rows.Next() { + rows.Scan(&c.ID, &c.Name, &c.Description, &c.OrgAppName, + &c.Created, &c.CreatedBy, &c.Updated, &c.UpdatedBy, + &c.ChangeSelector) + dcs = append(dcs, c) + } + + Expect(len(dcs)).To(Equal(1)) + dc := dcs[0] + + Expect(dc.ID).To(Equal("i")) + Expect(dc.Name).To(Equal("n")) + Expect(dc.Description).To(Equal("d")) + Expect(dc.OrgAppName).To(Equal("o")) + Expect(dc.Created).To(Equal("c")) + Expect(dc.CreatedBy).To(Equal("c")) + Expect(dc.Updated).To(Equal("u")) + Expect(dc.UpdatedBy).To(Equal("u")) + Expect(dc.ChangeSelector).To(Equal("c")) + + // Data Scope + var dds []dataDataScope + + rows, err = db.Query(` + SELECT id, apid_cluster_id, scope, org, + env, created, created_by, updated, + updated_by, _change_selector + FROM DATA_SCOPE`) + Expect(err).NotTo(HaveOccurred()) + defer rows.Close() + + d := dataDataScope{} + for rows.Next() { + rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org, + &d.Env, &d.Created, &d.CreatedBy, &d.Updated, + &d.UpdatedBy, &d.ChangeSelector) + dds = append(dds, d) + } + + Expect(len(dds)).To(Equal(1)) + ds := dds[0] + + Expect(ds.ID).To(Equal("i")) + Expect(ds.Org).To(Equal("o")) + Expect(ds.Env).To(Equal("e")) + Expect(ds.Scope).To(Equal("s")) + Expect(ds.Created).To(Equal("c")) + Expect(ds.CreatedBy).To(Equal("c")) + Expect(ds.Updated).To(Equal("u")) + Expect(ds.UpdatedBy).To(Equal("u")) + Expect(ds.ChangeSelector).To(Equal("c")) + }) + }) + + Context("ApigeeSync change event", func() { + + Context(LISTENER_TABLE_APID_CLUSTER, func() { + + It("insert event should panic", func() { + + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Insert, + Table: LISTENER_TABLE_APID_CLUSTER, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + It("update event should panic", func() { + + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Update, + Table: LISTENER_TABLE_APID_CLUSTER, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + PIt("delete event should kill all the things!") + }) + + Context(LISTENER_TABLE_DATA_SCOPE, func() { + + It("insert event should add", func() { + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Insert, + Table: LISTENER_TABLE_DATA_SCOPE, + NewRow: 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) + + var dds []dataDataScope + + rows, err := getDB().Query(` + SELECT id, apid_cluster_id, scope, org, + env, created, created_by, updated, + updated_by, _change_selector + FROM DATA_SCOPE`) + Expect(err).NotTo(HaveOccurred()) + defer rows.Close() + + d := dataDataScope{} + for rows.Next() { + rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org, + &d.Env, &d.Created, &d.CreatedBy, &d.Updated, + &d.UpdatedBy, &d.ChangeSelector) + dds = append(dds, d) + } + + Expect(len(dds)).To(Equal(1)) + ds := dds[0] + + Expect(ds.ID).To(Equal("i")) + Expect(ds.Org).To(Equal("o")) + Expect(ds.Env).To(Equal("e")) + Expect(ds.Scope).To(Equal("s")) + Expect(ds.Created).To(Equal("c")) + Expect(ds.CreatedBy).To(Equal("c")) + Expect(ds.Updated).To(Equal("u")) + Expect(ds.UpdatedBy).To(Equal("u")) + Expect(ds.ChangeSelector).To(Equal("c")) + }) + + It("delete event should delete", func() { + insert := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Insert, + Table: LISTENER_TABLE_DATA_SCOPE, + NewRow: 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(&insert) + + delete := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Delete, + Table: LISTENER_TABLE_DATA_SCOPE, + OldRow: insert.Changes[0].NewRow, + }, + }, + } + + handler.Handle(&delete) + + var nRows int + err := getDB().QueryRow("SELECT count(id) FROM DATA_SCOPE").Scan(&nRows) + Expect(err).NotTo(HaveOccurred()) + + Expect(nRows).To(Equal(0)) + }) + + It("update event should panic", func() { + + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Update, + Table: LISTENER_TABLE_DATA_SCOPE, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + }) + + }) +})
diff --git a/payload.go b/payload.go deleted file mode 100644 index 1728922..0000000 --- a/payload.go +++ /dev/null
@@ -1,55 +0,0 @@ -package apidApigeeSync - -type Payload struct { - Email string `json:"email"` - FirstName string `json:"firstName"` - LastName string `json:"lastName"` - UserName string `json:"userName"` - Organization string `json:"organizationName"` - Status string `json:"status"` - CreatedAt int64 `json:"createdAt"` - CreatedBy string `json:"createdBy"` - LastModifiedAt int64 `json:"lastModifiedAt"` - LastModifiedBy string `json:"lastModifiedBy"` - AppId string `json:"appId"` - AppFamily string `json:"appFamily"` - ConsumerSecret string `json:"consumerSecret"` - IssuedAt int64 `json:"issuedAt"` - DeveloperId string `json:"developerId"` - CallbackUrl string `json:"callbackUrl"` - AppName string `json:"name"` - ApiProducts []Apip `json:"apiProducts"` - Environments []string `json:"environments"` - Resources []string `json:"apiResources"` - URL string `json:"url"` - Type int `json:"type"` - ParentId string `json:"parentId"` - Etag string `json:"etag"` - Customtag string `json:"customtag"` - Manifest string `json:"manifest"` -} - -type DataPayload struct { - EntityIdentifier string `json:"entityIdentifier"` - EntityType string `json:"entityType"` - Operation string `json:"operation"` - PldCont Payload `json:"entityPayload"` -} - -type ChangePayload struct { - LastMsId int64 `json:"_id"` - Ts int64 `json:"_ts"` - Tags []string `json:"tags"` - Data DataPayload `json:"data"` -} - -type ChangeSet struct { - AtStart bool `json:"atStart"` - AtEnd bool `json:"atEnd"` - Changes []ChangePayload `json:"changes"` -} - -type Apip struct { - ApiProduct string `json:"apiproduct"` - Status string `json:"status"` -}