Make apid work even if snapshot server is down, but has some valid data. Use Prepare/Exec for SQL Inserts.
diff --git a/apigee_sync.go b/apigee_sync.go index 605ef48..4aec8c2 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -68,10 +68,6 @@ } changesUri.Path = path.Join(changesUri.Path, "/changes") - /* - * FIXME: This is a hack, while the correct procedure it to use the - * bootstrap scope - */ configId := config.GetString(configScopeId) for { @@ -83,18 +79,26 @@ return errors.New("Unable to get new token") } } + /* Find the scopes associated with the config id */ scopes := findScopesforId(configId) - - /* A Blocking call for 1 Minute */ v := url.Values{} + + /* Sequence added to the query if available */ if gotSequence == true { v.Add("since", lastSequence) } v.Add("block", "60") + + /* + * Include all the scopes associated with the config Id + * The Config Id is included as well, as it acts as the + * Bootstrap scope + */ for _, scope := range scopes { v.Add("scope", scope) } + v.Add("scope", configId) v.Add("snapshot", snapshotInfo) changesUri.RawQuery = v.Encode() uri := changesUri.String() @@ -130,6 +134,7 @@ return err } + /* If valid data present, Emit to plugins */ if len(resp.Changes) > 0 { events.Emit(ApigeeSyncEventSelector, &resp) } else { @@ -210,9 +215,18 @@ return nil } +/* Method downloads the snapshot in a two phased manner. + * Phase 1: Use the apidConfigId as the bootstrap scope, and + * get the apid_config and apid_config_scope from the snapshot + * server. + * Phase 2: Get all the scopes fetches from phase 1, and issue + * the second call to the snapshot server to get all the data + * associated with the scope(s). + * Emit the data for the necessary plugins to process. + */ func DownloadSnapshot() error { -RETRY: +PHASE_2: var scopes []string /* Get the bearer token */ @@ -247,9 +261,31 @@ } req, err := http.NewRequest("GET", uri, nil) req.Header.Add("Authorization", "Bearer "+token) + + /* Set the transport protocol type based on conf file input */ + if config.GetString(configSnapshotProtocol) == "json" { + req.Header.Set("Accept", "application/json") + } else { + req.Header.Set("Accept", "application/proto") + } + + /* Issue the request to the snapshot server */ r, err := client.Do(req) if err != nil { - log.Fatalf("Snapshotserver comm error: [%s] ", err) + + /* + * Check if there is some data already, if there is, + * Proceed, else exit the service (Scopes & snapshotInfo + * have to be present for changeserver to function) + */ + snapshotInfo = findSnapshotInfo(config.GetString(configScopeId)) + if snapshotInfo == "" { + log.Fatalf("Snapshotserver comm error: [%s] ", err) + } else { + log.Error("Snapshot server down, Proceeding with local data") + downloadSnapshot = true + return nil + } } defer r.Body.Close() @@ -264,7 +300,7 @@ /* * The idea here is that you download snapshot for the scopes * associated with the apidconfig Id, and then download the - * data based on the scopes retrieved in the first round + * data based on the scopes retrieved in the first phase */ if r.StatusCode == 200 { log.Info("Emit Snapshot response to plugins") @@ -272,7 +308,7 @@ snapshotInfo = resp.SnapshotInfo if downloadBootSnapshot == false { downloadBootSnapshot = true - goto RETRY + goto PHASE_2 } else if downloadBootSnapshot == true { downloadSnapshot = true } @@ -283,6 +319,10 @@ return err } +/* + * For the given apidConfigId, this function will retrieve all the scopes + * associated with it + */ func findScopesforId(configId string) (scopes []string) { var scope string @@ -304,3 +344,26 @@ } return scopes } + +/* + * Retrieve SnapshotInfo for the given apidConfigId from apid_config table + */ +func findSnapshotInfo(configId string) (snapshotInfo string) { + + db, err := data.DB() + if err != nil { + log.Errorf("DB open Error: %s", err) + return "" + } + + rows, err := db.Query("select snapshotInfo from APID_CONFIG where id = $1", configId) + if err != nil { + log.Errorf("Failed to query APID_CONFIG. Err: %s", err) + return "" + } + defer rows.Close() + for rows.Next() { + rows.Scan(&snapshotInfo) + } + return snapshotInfo +}
diff --git a/init.go b/init.go index 81bc050..506ffe7 100644 --- a/init.go +++ b/init.go
@@ -14,8 +14,9 @@ configConsumerKey = "apigeesync_consumer_key" configConsumerSecret = "apigeesync_consumer_secret" configScopeId = "apigeesync_bootstrap_id" - - ApigeeSyncEventSelector = "ApigeeSync" + configSnapshotProtocol = "apigeesync_snapshot_proto" + configUnitTestMode = "apigeesync_UnitTest_mode" + ApigeeSyncEventSelector = "ApigeeSync" ) var ( @@ -86,6 +87,7 @@ updated int64, updated_by text, _apid_scope text, + snapshotInfo text, PRIMARY KEY (id) ); CREATE TABLE apid_config_scope (
diff --git a/listener.go b/listener.go index 3730e14..36be24d 100644 --- a/listener.go +++ b/listener.go
@@ -45,7 +45,7 @@ switch payload.Name { case "apid_config": for _, row := range payload.Rows { - insertApidConfig(row, db) + insertApidConfig(row, db, snapshot.SnapshotInfo) } case "apid_config_scope": for _, row := range payload.Rows { @@ -67,13 +67,6 @@ for _, payload := range changes.Changes { switch payload.Table { - case "public.apid_config": - case "edgex.apid_config": - switch payload.Operation { - case 1: - insertApidConfig(payload.NewRow, db) - } - case "public.apid_config_scope": case "edgex.apid_config_scope": switch payload.Operation { @@ -87,23 +80,27 @@ /* * INSERT INTO APP_CREDENTIAL op */ -func insertApidConfig(ele common.Row, db *sql.DB) { +func insertApidConfig(ele common.Row, db *sql.DB, snapInfo string) bool { var scope, id, name, orgAppName, createdBy, updatedBy, Description string var updated, created int64 - txn, _ := db.Begin() - err := ele.Get("id", &id) - err = ele.Get("_apid_scope", &scope) - err = ele.Get("name", &name) - err = ele.Get("umbrella_org_app_name", &orgAppName) - err = ele.Get("created", &created) - err = ele.Get("created_by", &createdBy) - err = ele.Get("updated", &updated) - err = ele.Get("updated_by", &updatedBy) - err = ele.Get("description", &Description) + ele.Get("id", &id) + ele.Get("_apid_scope", &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) - _, err = txn.Exec("INSERT INTO apid_config (id, _apid_scope, name, umbrella_org_app_name, created, created_by, updated, updated_by)VALUES(?,?,?,?,?,?,?,?);", + stmt, err := db.Prepare("INSERT INTO APID_CONFIG (id, _apid_scope, name, umbrella_org_app_name, created, created_by, updated, updated_by, snapshotInfo)VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9);") + if err != nil { + log.Error("INSERT APID_CONFIG Failed: ", err) + return false + } + _, err = stmt.Exec( id, scope, name, @@ -112,14 +109,14 @@ createdBy, updated, updatedBy, - Description) + snapInfo) if err != nil { - log.Error("INSERT Failed: ", id, ", ", scope, ")", err) - txn.Rollback() + log.Error("INSERT APID_CONFIG Failed: ", id, ", ", scope, ")", err) + return false } else { - log.Info("INSERT Success: (", id, ", ", scope, ")") - txn.Commit() + log.Info("INSERT APID_CONFIG Success: (", id, ", ", scope, ")") + return true } } @@ -127,22 +124,23 @@ /* * INSERT INTO APP_CREDENTIAL op */ -func insertApidConfigScope(ele common.Row, db *sql.DB) { +func insertApidConfigScope(ele common.Row, db *sql.DB) bool { var id, scopeId, apiConfigId, scope, createdBy, updatedBy string var created, updated int64 - txn, _ := db.Begin() - err := ele.Get("id", &id) - err = ele.Get("_apid_scope", &scopeId) - err = ele.Get("apid_config_id", &apiConfigId) - err = ele.Get("scope", &scope) - err = ele.Get("created", &created) - err = ele.Get("created_by", &createdBy) - err = ele.Get("updated", &updated) - err = ele.Get("updated_by", &updatedBy) + ele.Get("id", &id) + ele.Get("_apid_scope", &scopeId) + ele.Get("apid_config_id", &apiConfigId) + ele.Get("scope", &scope) + ele.Get("created", &created) + ele.Get("created_by", &createdBy) + ele.Get("updated", &updated) + ele.Get("updated_by", &updatedBy) - _, err = txn.Exec("INSERT INTO apid_config_scope (id, _apid_scope, apid_config_id, scope, created, created_by, updated, updated_by)VALUES(?,?,?,?,?,?,?,?);", + stmt, err := db.Prepare("INSERT INTO APID_CONFIG_SCOPE (id, _apid_scope, apid_config_id, scope, created, created_by, updated, updated_by)VALUES($1,$2,$3,$4,$5,$6,$7,$8);") + + _, err = stmt.Exec( id, scopeId, apiConfigId, @@ -153,11 +151,11 @@ updatedBy) if err != nil { - log.Error("INSERT CRED Failed: ", id, ", ", scope, ")", err) - txn.Rollback() + log.Error("INSERT APID_CONFIG_SCOPE Failed: ", id, ", ", scope, ")", err) + return false } else { - log.Info("INSERT CRED Success: (", id, ", ", scope, ")") - txn.Commit() + log.Info("INSERT APID_CONFIG_SCOPE Success: (", id, ", ", scope, ")") + return true } }