snapshot db protocol: emit snapshot on existing db start and document protocol in readme, cleanup
diff --git a/README.md b/README.md index 8532ecb..a274032 100644 --- a/README.md +++ b/README.md
@@ -1,38 +1,109 @@ # apidApigeeSync This core plugin for [apid](http://github.com/30x/apid) connects to the Apigee Change Agent and publishes the data -changes events onto the apid Event service. +changes events onto the apid Event service. It also coordinates DB initialization for plugins on startup. ### Configuration -#### apigeesync_poll_interval +| name | description | +|------------------------------|--------------------------| +| apigeesync_poll_interval | int. seconds. default: 5 | +| apigeesync_organization | string. name. required. | +| apigeesync_proxy_server_base | string. url. required. | +| apigeesync_consumer_key | string. required. | +| apigeesync_consumer_secret | string. required. | -int. seconds. default: 5 +### Event Generated -#### apigeesync_organization +* Selector: "ApigeeSync" +* Data: [payload.go](payload.go) -string. name. required. +### Startup Procedure -#### apigeesync_proxy_server_base +#### ApigeeSync +1. Read DB version (Snapshot.SnapshotInfo) from default DB +2. If version found, emit Snapshot event (using Snapshot.SnapshotInfo, no data) +3. Ask server for Snapshot +4. Each time a Snapshot is received + 1. Verify Snapshot.SnapshotInfo is different than current + 2. Stop processing change events + 3. Remove or clean new DB version if it exists + 4. Emit Snapshot event + 5. Wait for plugins to finish processing + 6. Save Snapshot.SnapshotInfo in default DB + 7. Release old DB version + 8. Start processing change events -string. url. required. +ToDo: ApigeeSync currently only receives a new snapshot during startup, so step #4 only happens once. However, it + will eventually receive snapshots over time and the sub-steps should be followed at that time. Plugins + depending on ApigeeSync for data should assume that it can happen at any time and follow the heuristic below. -#### apigeesync_consumer_key +#### ApigeeSync-dependent plugins +1. Initialization + 1. Until receiving first Snapshot message, ApigeeSync-dependent APIs must either: + 1. not register (endpoint will return a 404 by default) + 2. return a 503 until DB is initialized +2. Upon receiving a snapshot notification (this is a HOT DB upgrade) + 1. Get DB for version (use Snapshot.SnapshotInfo as version) + 2. Create plugin's tables, if needed + 3. Insert any snapshot data into plugin's tables + 4. Set reference to new DB for all data access + 5. If db-dependent services are not exposed yet, expose them -string. required. +Example plugin code: -#### apigeesync_consumer_secret + var ( + log apid.LogService // set in initPlugin + data apid.DataService + unsafeDB apid.DB + dbMux sync.RWMutex + ) + + func init() { + apid.RegisterPlugin(initPlugin) + } + + func initPlugin(services apid.Services) error { + log = services.Log().ForModule("examplePlugin") + log.Debug("start init") + data = services.Data() + return nil + } + + // always use getDB() to safely access current DB + func getDB() apid.DB { + dbMux.RLock() + db := unsafeDB + dbMux.RUnlock() + return db + } + + func setDB(db apid.DB) { + dbMux.Lock() + if unsafeDB == nil { // init API on DB initialization (optional) + go initAPI() + } + unsafeDB = db + dbMux.Unlock() + } -string. required. + func processSnapshot(snapshot *common.Snapshot) { + + log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo) + + db, err := data.DBVersion(snapshot.SnapshotInfo) + if err != nil { + log.Panicf("Unable to access database: %v", err) + } - -### Event - -Selector: - - apidApigeeSync.ApigeeSyncEventSelector - - -Data: - -See: [payload.go](payload.go) + // init DB as needed (note: DB may exist, use 'CREATE TABLE IF NOT EXISTS' if not explicitly checking) + initDB(db) + + for _, table := range snapshot.Tables { + // populate tables from snapshot... + } + + // switch to new database + setDB(db) + log.Debug("Snapshot processed") + }
diff --git a/apigee_sync.go b/apigee_sync.go index 8184353..e02a790 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -13,9 +13,6 @@ "time" ) -// todo: The following was largely copied from old APID - needs review - -var latestSequence int64 var token string var tokenActive, downloadDataSnapshot, downloadBootSnapshot, chfin bool var lastSequence string @@ -270,7 +267,7 @@ * 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. - * If there is already previous data in sqlite, donot fetch + * If there is already previous data in sqlite, don't fetch * again from snapshot server. */ func DownloadSnapshots() { @@ -283,6 +280,21 @@ if gsnapshotInfo != "" { downloadDataSnapshot = true downloadBootSnapshot = true + + log.Infof("Starting on downloaded snapshot: %s", gsnapshotInfo) + + // verify DB is accessible + _, err := data.DBVersion(gsnapshotInfo) + if err != nil { + log.Panicf("Database inaccessible: %v", err) + } + + // allow plugins to start immediately on existing database + snap := &common.Snapshot{ + SnapshotInfo: gsnapshotInfo, + } + events.Emit(ApigeeSyncEventSelector, snap) + return } @@ -294,7 +306,7 @@ * Snapshot */ for count := 0; count < 60; count++ { - if downloadBootSnapshot == false { + if !downloadBootSnapshot { log.Debug("Waiting for bootscope snapshot download...") time.Sleep(time.Duration(count) * 100 * time.Millisecond) } else { @@ -303,7 +315,7 @@ } /* Phase 2 */ - if downloadBootSnapshot == true && downloadDataSnapshot == true { + if downloadBootSnapshot && downloadDataSnapshot { log.Debug("Proceeding with existing Sqlite data") } else if downloadBootSnapshot == true { log.Debug("Proceed to download Snapshot for data scopes") @@ -423,7 +435,7 @@ } /* - * Retrieve SnapshotInfo for the given apidConfigId from apid_config table + * Retrieve LastSequence for the given apidConfigId from apid_config table */ func findLastSeqInfo(configId string) (info string) { @@ -446,7 +458,7 @@ } /* - * Retrieve LastSequence for the given apidConfigId from apid_config table + * Retrieve SnapshotInfo for the given apidConfigId from apid_config table */ func findSnapshotInfo(configId string) (info string) {
diff --git a/init.go b/init.go index 3edc842..65b72ad 100644 --- a/init.go +++ b/init.go
@@ -57,7 +57,7 @@ /* 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 intialized + * after all the plugins are initialized */ events.ListenFunc(apid.SystemEventsSelector, postInitPlugins)
diff --git a/listener.go b/listener.go index 83b87dd..32a1d94 100644 --- a/listener.go +++ b/listener.go
@@ -36,7 +36,7 @@ if ok { res = processChange(changeSet, txn) } else { - log.Fatal("Received Invalid event. This shouldn't happen!") + log.Fatal("Received invalid event: %v", e) } } if res == true {