diff --git a/Dockerfile b/Dockerfile index f9c12239..62fa151f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -ARG NODE_VERSION=16.20.2-bullseye-slim +ARG NODE_VERSION=16.20.2-bookworm-slim # Use separate builder to retrieve & build node modules FROM node:${NODE_VERSION} AS builder @@ -34,6 +34,7 @@ RUN apt-get update && \ jq \ python3 \ python3-pip \ + python3-venv \ && rm -rf /var/lib/apt/lists/* ENV BALLOT_VERSION 1.0.4 @@ -48,6 +49,8 @@ COPY --from=builder /usr/src/app/supervisord /usr/local/bin/ ENV NO_PROXY localhost,127.0.0.1 ENV no_proxy localhost,127.0.0.1 +RUN python3 -m venv /opt/venv +ENV PATH="/opt/venv/bin:$PATH" RUN pip install -r utapi/requirements.txt ## This section duplicates S3C Federation Dockerfile, this needs to be refactored diff --git a/ObjectRepair/DuplicateKeysIngestion.js b/ObjectRepair/DuplicateKeysIngestion.js index c90c0293..4d6cb51e 100644 --- a/ObjectRepair/DuplicateKeysIngestion.js +++ b/ObjectRepair/DuplicateKeysIngestion.js @@ -1,5 +1,6 @@ const { waterfall } = require('async'); const { Logger } = require('werelogs'); +const { errors } = require('arsenal'); const { httpRequest } = require('../repairDuplicateVersionsSuite'); const { SproxydKeysProcessor } = require('./DuplicateKeysWindow'); const { ProxyLoggerCreator, AggregateLogger } = require('./Logging'); @@ -73,10 +74,26 @@ class RaftJournalReader { log.error('unable to fetch cseq', { err, requestUrl }); return cb(err); } + if (res.statusCode === 416) { + // RAFT session is empty: start at 1 + this.cseq = 1; + } else { + if (res.statusCode !== 200) { + log.error( + 'unable to fetch cseq: error HTTP status code returned', + { statusCode: res.statusCode }, + ); + return cb(errors.InternalError); + } - const body = JSON.parse(res.body); - - this._setCseq(body); + try { + const body = JSON.parse(res.body); + this._setCseq(body); + } catch (err) { + log.error('unable to fetch cseq: invalid JSON in response body'); + return cb(errors.InternalError); + } + } // make sure begin is at least 1 since Raft Journal logs are 1-indexed this.begin = Math.max(1, this.cseq - this.lookBack); @@ -134,7 +151,12 @@ class RaftJournalReader { return cb(new Error(`GET ${requestUrl} returned empty body at ${this.begin}`)); } - const body = JSON.parse(res.body); + let body; + try { + body = JSON.parse(res.body); + } catch (err) { + return cb(new Error(`GET ${requestUrl}: invalid JSON in response body`)); + } // FIXME this special case should be taken care of once S3C-3928 is fixed if (body.log.length === 0) { return cb(new RangeError(`GET ${requestUrl} found no new records at ${this.begin}`)); diff --git a/README.md b/README.md index b0f9dee2..eabfa82b 100644 --- a/README.md +++ b/README.md @@ -1512,6 +1512,24 @@ and ignores other entries. cat /tmp/verifyBucketSproxydKeys.log | docker run -i zenko/s3utils:latest bash -c 'OBJECT_REPAIR_BUCKETD_HOSTPORT=127.0.0.1:9000 OBJECT_REPAIR_SPROXYD_HOSTPORT=127.0.0.1:8181 node repairDuplicateVersions.js' > /tmp/repairDuplicateVersions.log ``` +## Effect on object metadata + +In addition to updating the `location` field, the repair tool adds a +`repairedInfo` field to the metadata of every repaired object version: + +```json +"repairedInfo": { + "lastRepairedAt": "2026-06-17T10:30:00.000Z", + "reason": "duplicate-sproxyd-keys" +} +``` + +This field serves as a forensic marker. `lastRepairedAt` is an ISO +8601 timestamp of when the repair was performed; `reason` identifies +the type of inconsistency that triggered the repair. If an object is +repaired more than once, the field is overwritten with the information +from the most recent repair. + ## Caveat While the script is running, there is a possibility, although slim, diff --git a/package.json b/package.json index e0c73be7..e8a514ea 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "s3utils", - "version": "1.14.19", + "version": "1.14.19-1", "engines": { "node": ">= 16" }, diff --git a/repairDuplicateVersionsSuite.js b/repairDuplicateVersionsSuite.js index 82561fb0..7bfab0fb 100644 --- a/repairDuplicateVersionsSuite.js +++ b/repairDuplicateVersionsSuite.js @@ -21,6 +21,7 @@ const { OBJECT_REPAIR_TLS_KEY_PATH, OBJECT_REPAIR_TLS_CERT_PATH, OBJECT_REPAIR_TLS_CA_PATH, + OBJECT_REPAIR_SOCKET_TIMEOUT_SECONDS, } = process.env; const useHttps = (OBJECT_REPAIR_TLS_KEY_PATH !== undefined @@ -42,9 +43,11 @@ const bucketdAgent = useHttps ? [fs.readFileSync(OBJECT_REPAIR_TLS_CA_PATH)] : undefined, keepAlive: true, + timeout: (Number(OBJECT_REPAIR_SOCKET_TIMEOUT_SECONDS) || 300) * 1000, }) : new httpArsn.Agent({ keepAlive: true, + timeout: (Number(OBJECT_REPAIR_SOCKET_TIMEOUT_SECONDS) || 300) * 1000, }); let sproxydAlias; @@ -99,9 +102,15 @@ function httpRequest(method, url, reqBody, cb) { `error sending HTTP request to ${url}: ${err.message}` ))); if (reqBody) { + let reqBuffer; + if (Buffer.isBuffer(reqBody)) { + reqBuffer = reqBody; + } else { + reqBuffer = Buffer.from(reqBody, 'utf8'); + } req.setHeader('content-type', 'application/json'); - req.setHeader('content-length', reqBody.length); - req.write(reqBody); + req.setHeader('content-length', reqBuffer.length); + req.write(reqBuffer); } req.end(); } @@ -371,6 +380,12 @@ function repairObject(objInfo, cb) { loc.key = copiedKeys[loc.key]; } }); + // Add some info in the metadata to help forensics, if only to + // mark that the metadata has been repaired. + objMD.repairedInfo = { + lastRepairedAt: new Date().toISOString(), + reason: 'duplicate-sproxyd-keys', + }; return putObjectMetadata(objInfo.objectUrl, objMD, err => { if (err) { log.error('error putting object metadata', { diff --git a/tests/unit/SproxydKeysScan/DuplicateKeysIngestionTest.js b/tests/unit/SproxydKeysScan/DuplicateKeysIngestionTest.js index 9282e0e3..58983216 100644 --- a/tests/unit/SproxydKeysScan/DuplicateKeysIngestionTest.js +++ b/tests/unit/SproxydKeysScan/DuplicateKeysIngestionTest.js @@ -1,13 +1,18 @@ const fs = require('fs'); +const { errors } = require('arsenal'); const { RaftJournalReader } = require('../../../ObjectRepair/DuplicateKeysIngestion'); const { subscribers } = require('../../../ObjectRepair/SproxydKeysSubscribers'); function getMockResponse(mockStatusCode) { - const mockBody = fs.readFileSync(`${__dirname}/RaftJournalTestData.json`, 'utf8'); const mockResponse = { - body: mockBody, statusCode: mockStatusCode, }; + if (Math.floor(mockStatusCode / 100) === 2) { + const mockBody = fs.readFileSync(`${__dirname}/RaftJournalTestData.json`, 'utf8'); + mockResponse.body = mockBody; + } else { + mockResponse.body = ''; + } return mockResponse; } @@ -39,35 +44,36 @@ describe('RaftJournalReader', () => { describe('::setBegin', () => { const reader = setupJournalReader(); - beforeEach(() => { - reader._httpRequest = mockHttpRequest(null, getMockResponse(200)); - }); - afterEach(() => { reader._httpRequest.mockReset(); }); - test('when begin is undefined, begin = latest cseq - lookBack', () => { + test('when begin is undefined, begin = latest cseq - lookBack', done => { + reader._httpRequest = mockHttpRequest(null, getMockResponse(200)); reader.begin = undefined; reader.lookBack = 5; reader.setBegin(err => { expect(err).toBe(undefined); expect(reader.begin).toEqual(reader.cseq - reader.lookBack); + done(); }); }); - test('begin is at minimum 1, even with lookback > latest cseq', () => { + test('begin is at minimum 1, even with lookback > latest cseq', done => { + reader._httpRequest = mockHttpRequest(null, getMockResponse(200)); reader.begin = undefined; reader.lookBack = Infinity; reader.setBegin(err => { expect(err).toBe(undefined); expect(reader.begin).toEqual(1); + done(); }); }); - test('if begin is set from a previous call, use the existing begin and ignore lookBack', () => { + test('if begin is set from a previous call, use the existing begin and ignore lookBack', done => { + reader._httpRequest = mockHttpRequest(null, getMockResponse(200)); reader.begin = 3; reader.lookBack = Infinity; @@ -75,6 +81,51 @@ describe('RaftJournalReader', () => { expect(err).toBe(undefined); expect(reader._httpRequest).not.toHaveBeenCalled(); expect(reader.begin).toEqual(3); + done(); + }); + }); + + test('when begin is undefined and route returns an unexpected error', done => { + reader._httpRequest = mockHttpRequest(new Error('OOPS'), undefined); + reader.begin = undefined; + + reader.setBegin(err => { + expect(err).toEqual(new Error('OOPS')); + done(); + }); + }); + + test('when begin is undefined and route returns a 416 HTTP status with empty response body', done => { + reader._httpRequest = mockHttpRequest(null, getMockResponse(416)); + reader.begin = undefined; + + reader.setBegin(err => { + expect(err).toBe(undefined); + expect(reader.begin).toEqual(1); + done(); + }); + }); + + test('when begin is undefined and route returns a 500 HTTP status with empty response body', done => { + reader._httpRequest = mockHttpRequest(null, getMockResponse(500)); + reader.begin = undefined; + + reader.setBegin(err => { + expect(err).toEqual(errors.InternalError); + done(); + }); + }); + + test('when begin is undefined and route returns a 200 OK HTTP status with invalid JSON in response body', done => { + reader._httpRequest = mockHttpRequest(null, { + statusCode: 200, + body: '{BADJSON}', + }); + reader.begin = undefined; + + reader.setBegin(err => { + expect(err).toEqual(errors.InternalError); + done(); }); }); }); @@ -96,7 +147,7 @@ describe('RaftJournalReader', () => { expect(body).toBe(undefined); }; - test('should correctly read mocked data', () => { + test('should correctly read mocked data', done => { reader._httpRequest = mockHttpRequest(null, getMockResponse(200)); reader.getBatch((err, body) => { expect(reader._httpRequest).toHaveBeenCalled(); @@ -109,35 +160,51 @@ describe('RaftJournalReader', () => { expect(err).toBe(null); expect(body).not.toBe(null); expect(body.log).not.toBe(null); + done(); }); }); - test('should return error with a non-200 status code', () => { + test('should return error with a non-200 status code', done => { reader._httpRequest = mockHttpRequest(null, getMockResponse(500)); reader.getBatch((err, body) => { returnedError(err, body); + done(); }); }); - test('should return error with a missing body', () => { + test('should return error with a missing body', done => { reader._httpRequest = mockHttpRequest(null, { statusCode: 200 }); reader.getBatch((err, body) => { returnedError(err, body); + done(); }); }); - test('should return error with null response', () => { + test('should return error with null response', done => { reader._httpRequest = mockHttpRequest(null, null); reader.getBatch((err, body) => { returnedError(err, body); + done(); }); }); // FIXME this special case should be taken care of once S3C-3928 is fixed - test('should return error with response containing empty log', () => { + test('should return error with response containing empty log', done => { reader._httpRequest = mockHttpRequest(null, getEmptyRsMockResponse()); reader.getBatch((err, body) => { returnedError(err, body); + done(); + }); + }); + + test('should return error with response containing invalid JSON', done => { + reader._httpRequest = mockHttpRequest(null, { + statusCode: 200, + body: '{BADJSON}', + }); + reader.getBatch((err, body) => { + returnedError(err, body); + done(); }); }); }); @@ -211,7 +278,7 @@ describe('RaftJournalReader', () => { reader._httpRequest = mockHttpRequest(null, getMockResponse(200)); afterEach(() => reader._httpRequest.mockReset()); - test('inserts correct sproxyd key data into SproxydKeyProcessor', () => { + test('inserts correct sproxyd key data into SproxydKeyProcessor', done => { expect(reader.begin).toEqual(1); reader.runOnce((err, timeout) => { expect(err).toBe(null); @@ -219,6 +286,7 @@ describe('RaftJournalReader', () => { expect(reader._httpRequest).toHaveBeenCalled(); expect(reader.processor.insert).toHaveBeenCalled(); expect(reader.begin).toBeGreaterThan(1); + done(); }); }); @@ -229,19 +297,21 @@ describe('RaftJournalReader', () => { expect(reader.processor.insert).not.toHaveBeenCalled(); }; - test('should set a timeout of 5000 milliseconds when there is any error during ingestion', () => { + test('should set a timeout of 5000 milliseconds when there is any error during ingestion', done => { reader._httpRequest = mockHttpRequest(null, getMockResponse(500)); reader.processor.insert.mockReset(); reader.runOnce((err, timeout) => { waitsFiveSeconds(err, timeout); + done(); }); }); - test('should set a timeout of 5000 milliseconds when there is no new data from raft journal', () => { + test('should set a timeout of 5000 milliseconds when there is no new data from raft journal', done => { reader._httpRequest = mockHttpRequest(null, null); reader.processor.insert.mockReset(); reader.runOnce((err, timeout) => { waitsFiveSeconds(err, timeout); + done(); }); }); }); diff --git a/tests/unit/repairDuplicateVersionsSuite.spec.js b/tests/unit/repairDuplicateVersionsSuite.spec.js new file mode 100644 index 00000000..c816ea87 --- /dev/null +++ b/tests/unit/repairDuplicateVersionsSuite.spec.js @@ -0,0 +1,143 @@ +const http = require('http'); +const { httpRequest } = require('../../repairDuplicateVersionsSuite'); + +describe('repairDuplicateVersionsSuite::httpRequest', () => { + let server; + let serverPort; + let lastRequest; + + beforeAll(done => { + server = http.createServer((req, res) => { + lastRequest = { + method: req.method, + path: req.url, + headers: req.headers, + body: '', + }; + req.on('data', chunk => { lastRequest.body += chunk.toString('utf8'); }); + req.on('end', () => { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true })); + }); + }); + server.listen(0, '127.0.0.1', () => { + serverPort = server.address().port; + done(); + }); + }); + + afterAll(done => { + server.close(done); + }); + + beforeEach(() => { + lastRequest = null; + }); + + test('GET returns response with body attached', done => { + const url = `http://127.0.0.1:${serverPort}/test-path`; + httpRequest('GET', url, null, (err, res) => { + expect(err).toBeNull(); + expect(res.statusCode).toBe(200); + expect(JSON.parse(res.body)).toEqual({ ok: true }); + expect(lastRequest.method).toBe('GET'); + expect(lastRequest.path).toBe('/test-path'); + done(); + }); + }); + + test('POST with multi-byte UTF-8 body sets content-length in bytes, not characters', done => { + // String length (13 chars) differs from byte length (19 bytes) due to + // multi-byte UTF-8 encoding of the Japanese characters and emoji. + const body = '{"key":"日本語🎉"}'; + const expectedByteLength = Buffer.byteLength(body, 'utf8'); + expect(expectedByteLength).toBeGreaterThan(body.length); + + const url = `http://127.0.0.1:${serverPort}/write-path`; + httpRequest('POST', url, body, (err, res) => { + expect(err).toBeNull(); + expect(res.statusCode).toBe(200); + expect(lastRequest.method).toBe('POST'); + expect(lastRequest.headers['content-type']).toBe('application/json'); + expect(parseInt(lastRequest.headers['content-length'], 10)).toBe(expectedByteLength); + expect(lastRequest.body).toBe(body); + done(); + }); + }); +}); + +describe('repairDuplicateVersionsSuite::putObjectMetadata', () => { + let putObjectMetadata; + let server; + let serverPort; + let lastRequest; + let responseStatus; + + beforeAll(done => { + server = http.createServer((req, res) => { + lastRequest = { + method: req.method, + path: req.url, + headers: req.headers, + body: '', + }; + req.on('data', chunk => { lastRequest.body += chunk.toString('utf8'); }); + req.on('end', () => { + res.writeHead(responseStatus); + res.end(); + }); + }); + server.listen(0, '127.0.0.1', () => { + serverPort = server.address().port; + // OBJECT_REPAIR_BUCKETD_HOSTPORT is captured at module load time, so + // set it before requiring the module via isolateModules. + process.env.OBJECT_REPAIR_BUCKETD_HOSTPORT = `127.0.0.1:${serverPort}`; + jest.isolateModules(() => { + ({ putObjectMetadata } = require('../../repairDuplicateVersionsSuite')); + }); + done(); + }); + }); + + afterAll(done => { + server.close(done); + }); + + beforeEach(() => { + lastRequest = null; + responseStatus = 200; + }); + + test('POSTs JSON-serialised metadata with multi-byte UTF-8 characters to the correct bucketd path', done => { + const md = { + 'content-type': 'text/plain', + 'size': 42, + 'key': '日本語🎉', + }; + putObjectMetadata('s3://mybucket/mykey', md, err => { + expect(err).toBeFalsy(); + expect(lastRequest.method).toBe('POST'); + expect(lastRequest.path).toBe('/default/bucket/mybucket/mykey'); + expect(JSON.parse(lastRequest.body)).toEqual(md); + done(); + }); + }); + + test('calls back with error immediately when objectUrl does not start with "s3://"', done => { + putObjectMetadata('http://mybucket/mykey', {}, err => { + expect(err).toBeInstanceOf(Error); + expect(err.message).toMatch(/malformed object URL/); + expect(lastRequest).toBeNull(); // no HTTP request made + done(); + }); + }); + + test('calls back with error when bucketd returns a non-200 status', done => { + responseStatus = 500; + putObjectMetadata('s3://mybucket/mykey', {}, err => { + expect(err).toBeInstanceOf(Error); + expect(err.message).toMatch(/500/); + done(); + }); + }); +});