ft: ZENKO-147 Use Redis keys instead of hash #286
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -12,6 +12,7 @@ const QueueEntry = require('../../lib/models/QueueEntry'); | |
| const Healthcheck = require('./Healthcheck'); | ||
| const routes = require('./routes'); | ||
| const { redisKeys } = require('../../extensions/replication/constants'); | ||
| const getFailedCRRKey = require('../util/getFailedCRRKey'); | ||
| const monitoringClient = require('../clients/monitoringHandler').client; | ||
|
|
||
| // StatsClient constant defaults | ||
|
|
@@ -216,57 +217,85 @@ class BackbeatAPI { | |
| /** | ||
| * Builds the failed CRR response. | ||
| * @param {String} cursor - The Redis HSCAN cursor | ||
| * @param {Array} keys - The collection of Redis keys for the iteration | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| _getFailedCRRResponse(cursor, keys, cb) { | ||
| const response = { | ||
| IsTruncated: Number.parseInt(cursor, 10) !== 0, | ||
| Versions: [], | ||
| }; | ||
| if (response.IsTruncated) { | ||
| response.NextMarker = Number.parseInt(cursor, 10); | ||
| } | ||
| const cmds = keys.map(k => ['get', k]); | ||
| return this._redisClient.batch(cmds, (err, res) => { | ||
| if (err) { | ||
| return cb(err); | ||
| } | ||
| for (let i = 0; i < res.length; i++) { | ||
| const [cmdErr, value] = res[i]; | ||
| if (cmdErr) { | ||
| return cb(cmdErr); | ||
| } | ||
| const queueEntry = QueueEntry.createFromKafkaEntry({ value }); | ||
| response.Versions.push({ | ||
| Bucket: queueEntry.getBucket(), | ||
| Key: queueEntry.getObjectKey(), | ||
| VersionId: queueEntry.getEncodedVersionId(), | ||
| StorageClass: queueEntry.getSite(), | ||
| Size: queueEntry.getContentLength(), | ||
| LastModified: queueEntry.getLastModified(), | ||
| }); | ||
| } | ||
| return cb(null, response); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Recursively scan all existing keys with a count of 1000. Call callback if | ||
| * the response is greater or equal to 1000 keys, or we have scanned all | ||
| * keys (i.e. when the cursor is 0). | ||
| * @param {String} pattern - The key pattern to match | ||
| * @param {Number} marker - The cursor to start scanning from | ||
| * @param {Array} allKeys - The collection of all matching keys found | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| _scanAllKeys(pattern, marker, allKeys, cb) { | ||
| const cmd = ['scan', marker, 'MATCH', pattern, 'COUNT', 1000]; | ||
| this._redisClient.batch([cmd], (err, res) => { | ||
| if (err) { | ||
| return cb(err); | ||
| } | ||
| const [cmdErr, collection] = res[0]; | ||
| if (cmdErr) { | ||
| return cb(cmdErr); | ||
| } | ||
| const [cursor, keys] = collection; | ||
| allKeys.push(...keys); | ||
| if (allKeys.length >= 1000 || Number.parseInt(cursor, 10) === 0) { | ||
| return cb(null, cursor, allKeys); | ||
| } | ||
| return this._scanAllKeys(pattern, cursor, allKeys, cb); | ||
|
There was a problem hiding this comment. I'm not familiar with redis API, but if asked for 1000 keys, may it return you less than this even if there are 1000+ keys to return? In such case it looks correct to call There was a problem hiding this comment. There isn't a guarantee on the number of keys being returned by Redis during the SCAN operation. (I've added a point about this in the Operational Considerations section of the doc for this feature.) We do have a guarantee that the scan has completed when the returned cursor is |
||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Find all failed CRR operations that match the bucket, key, and versionID. | ||
| * @param {Object} details - The route details | ||
| * @param {Function} cb - The callback to call | ||
| * @return {undefined} | ||
| */ | ||
| getFailedCRR(details, cb) { | ||
| const { bucket, key, versionId } = details; | ||
| const { failedCRR } = redisKeys; | ||
| const pattern = `${failedCRR}:${bucket}:${key}:${versionId}:*`; | ||
| return this._scanAllKeys(pattern, 0, [], (err, cursor, keys) => | ||
| this._getFailedCRRResponse(cursor, keys, cb)); | ||
| } | ||
|
|
||
| /** | ||
| * Get all CRR operations that have failed. | ||
| * @param {Object} details - The route details | ||
|
|
@@ -275,26 +304,17 @@ class BackbeatAPI { | |
| */ | ||
| getAllFailedCRR(details, cb) { | ||
| const marker = Number.parseInt(details.marker, 10) || 0; | ||
| const pattern = `${redisKeys.failedCRR}:*`; | ||
| return this._scanAllKeys(pattern, marker, [], (err, cursor, keys) => | ||
| this._getFailedCRRResponse(cursor, keys, cb)); | ||
| } | ||
|
|
||
| /** | ||
| * For the given queue entry's site, send an entry with PENDING status to | ||
| * the replication status topic, then send an entry to the replication topic | ||
| * so that the queue processor re-attempts replication. | ||
| * @param {QueueEntry} queueEntry - The queue entry constructed from the | ||
| * failed kafka entry that was stored as a Redis key value. | ||
| * @param {Function} cb - The callback. | ||
| * @return {undefined} | ||
| */ | ||
|
|
@@ -312,29 +332,27 @@ class BackbeatAPI { | |
| } | ||
|
|
||
| /** | ||
| * Delete the failed CRR Redis key. | ||
| * @param {String} key - The key to delete | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| _deleteFailedCRRField(key, cb) { | ||
| const cmd = ['del', key]; | ||
| return this._redisClient.batch([cmd], (err, res) => { | ||
| if (err) { | ||
| this._logger.error('error deleting redis key', { | ||
| method: 'BackbeatAPI._deleteFailedCRRField', | ||
| key, | ||
| error: err, | ||
| }); | ||
| return cb(err); | ||
| } | ||
| const [cmdErr] = res[0]; | ||
| if (cmdErr) { | ||
| this._logger.error('error deleting redis key', { | ||
| method: 'BackbeatAPI._deleteFailedCRRField', | ||
| key, | ||
| error: cmdErr, | ||
| }); | ||
| return cb(cmdErr); | ||
|
|
@@ -380,11 +398,12 @@ class BackbeatAPI { | |
| LastModified: queueEntry.getLastModified(), | ||
| ReplicationStatus: 'PENDING', | ||
| }); | ||
| const key = | ||
| getFailedCRRKey(Bucket, Key, VersionId, StorageClass); | ||
| return this._deleteFailedCRRField(key, err => { | ||
| if (err) { | ||
| this._logger.error('could not delete redis key after ' + | ||
| 'pushing to kafka topics', { | ||
| method: 'BackbeatAPI._processFailedKafkaEntries', | ||
| error: err, | ||
| }); | ||
|
|
@@ -408,20 +427,24 @@ class BackbeatAPI { | |
| if (error) { | ||
| return cb(error); | ||
| } | ||
| const cmds = reqBody.map(o => { | ||
| const { Bucket, Key, VersionId, StorageClass } = o; | ||
| const key = getFailedCRRKey(Bucket, Key, VersionId, StorageClass); | ||
| return ['get', key]; | ||
| }); | ||
| return this._redisClient.batch(cmds, (err, res) => { | ||
| if (err) { | ||
| return cb(err); | ||
| } | ||
| const entries = []; | ||
| for (let i = 0; i < res.length; i++) { | ||
| const [cmdErr, entry] = res[i]; | ||
| if (cmdErr) { | ||
| return cb(cmdErr); | ||
| } | ||
| entries.push(entry); | ||
| } | ||
| return this._processFailedKafkaEntries(entries, cb); | ||
| }); | ||
| } | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| const { redisKeys } = require('../../extensions/replication/constants'); | ||
|
|
||
| /** | ||
| * Returns the schema used for failed CRR entry Redis keys. | ||
| * @param {String} bucket - The name of the bucket | ||
| * @param {String} key - The name of the key | ||
| * @param {String} versionId - The encoded version ID | ||
| * @param {String} storageClass - The storage class of the object | ||
| * @return {String} - The Redis key used for the failed CRR entry | ||
| */ | ||
| function getFailedCRRKey(bucket, key, versionId, storageClass) { | ||
| const { failedCRR } = redisKeys; | ||
| return `${failedCRR}:${bucket}:${key}:${versionId}:${storageClass}`; | ||
| } | ||
|
|
||
| module.exports = getFailedCRRKey; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
allKeys.length >= 1000If we get 1000 keys initially, won't this return before getting next set of keys? Or was this intentional?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, in effect it allows for a paginated response. I wanted to limit the API response listing to ~1000 keys to model closer to the default for version listings in S3. We cannot guarantee the exact number because Redis does not make a guarantee for the number of keys returned during the scan. So if it's 1000 or more, the API will include a
NextMarkervalue for subsequent listings.