Compare commits
4 Commits
developmen
...
dev/ft/WIP
Author | SHA1 | Date |
---|---|---|
Antonin Coulibaly | 76a42ce0b6 | |
Antonin Coulibaly | 90f91d7d9d | |
Antonin Coulibaly | 6e75fbef25 | |
Antonin Coulibaly | e00f85ee57 |
|
@ -27,6 +27,7 @@
|
||||||
"host": "localhost",
|
"host": "localhost",
|
||||||
"port": 8500
|
"port": 8500
|
||||||
},
|
},
|
||||||
|
"kinetic": ["localhost:8123"],
|
||||||
"clusters": 10,
|
"clusters": 10,
|
||||||
"log": {
|
"log": {
|
||||||
"logLevel": "info",
|
"logLevel": "info",
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
import assert from 'assert';
|
import assert from 'assert';
|
||||||
import fs from 'fs';
|
import fs from 'fs';
|
||||||
import path from 'path';
|
import path from 'path';
|
||||||
|
import Kinetic from './data/kinetic/Kinetic';
|
||||||
import authDataChecker from './auth/in_memory/checker';
|
import authDataChecker from './auth/in_memory/checker';
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -279,6 +279,20 @@ class Config {
|
||||||
dataPath,
|
dataPath,
|
||||||
metadataPath,
|
metadataPath,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
if (config.kinetic && Array.isArray(config.kinetic)
|
||||||
|
&& process.env.S3SPROXYD === 'kinetic') {
|
||||||
|
this.kinetic = {};
|
||||||
|
this.kinetic.drives = [];
|
||||||
|
config.kinetic.forEach(drive => {
|
||||||
|
const tab = drive.split(':');
|
||||||
|
this.kinetic.drives.push({
|
||||||
|
host: tab[0],
|
||||||
|
port: tab[1],
|
||||||
|
});
|
||||||
|
});
|
||||||
|
this.kinetic.instance = new Kinetic(this.kinetic.drives);
|
||||||
|
}
|
||||||
return config;
|
return config;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,284 @@
|
||||||
|
import crypto from 'crypto';
|
||||||
|
import EventEmitter from 'events';
|
||||||
|
import net from 'net';
|
||||||
|
import stream from 'stream';
|
||||||
|
|
||||||
|
import { errors } from 'arsenal';
|
||||||
|
import kineticlib from 'kineticlib';
|
||||||
|
|
||||||
|
const HEADER_SZ = 9;
|
||||||
|
|
||||||
|
class Kinetic {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @constructor
|
||||||
|
* @param {Object.<string>[]} drives - the list of drives
|
||||||
|
* @returns {Kinetic} - to allow for a functional style.
|
||||||
|
*/
|
||||||
|
constructor(drives) {
|
||||||
|
this._events = new EventEmitter();
|
||||||
|
this._sockets = [];
|
||||||
|
this._drives = drives;
|
||||||
|
for (let i = 0; i < drives.length; i++) {
|
||||||
|
this._sockets.push(
|
||||||
|
{ sock: this._newSocket(),
|
||||||
|
connectionID: 0,
|
||||||
|
clusterVersion: 0,
|
||||||
|
drive: drives[i] });
|
||||||
|
}
|
||||||
|
this._request = false;
|
||||||
|
this._chunkTab = [];
|
||||||
|
this._count = 0;
|
||||||
|
this.connect();
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Connect the sockets to the drives
|
||||||
|
* @return {undefined}
|
||||||
|
*/
|
||||||
|
connect() {
|
||||||
|
let index = 0;
|
||||||
|
this._sockets.forEach(socket => {
|
||||||
|
socket.sock.on('connect', () => {
|
||||||
|
this._bindReadable(socket, index);
|
||||||
|
++index;
|
||||||
|
});
|
||||||
|
socket.sock.connect(socket.drive);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
_bindReadable(socket, index) {
|
||||||
|
socket.sock.on('readable', err => {
|
||||||
|
if (this._request === false) {
|
||||||
|
this._parsePDU(socket, (err1, pdu) => {
|
||||||
|
this._parseRequest(socket, err || err1, pdu, index);
|
||||||
|
});
|
||||||
|
} else {
|
||||||
|
this._endGet(socket);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
_parsePDU(socket, callback) {
|
||||||
|
const header = socket.sock.read(HEADER_SZ);
|
||||||
|
if (header !== null) {
|
||||||
|
const protobufSize = header.readInt32BE(1);
|
||||||
|
const rawData = socket.sock.read(protobufSize);
|
||||||
|
const pdu = new kineticlib.PDU(Buffer.concat([header, rawData]));
|
||||||
|
const err = this._propError(pdu);
|
||||||
|
return callback(err, pdu);
|
||||||
|
}
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
_parseRequest(socket, err, pdu, index) {
|
||||||
|
const sock = socket;
|
||||||
|
switch (pdu.getMessageType()) {
|
||||||
|
case null:
|
||||||
|
sock._initPDU = pdu;
|
||||||
|
sock.connectionID = pdu.getConnectionId();
|
||||||
|
sock.clusterVersion = pdu.getClusterVersion();
|
||||||
|
sock.index = index;
|
||||||
|
sock.sequence = 0;
|
||||||
|
socket.lexiKey = '0';
|
||||||
|
return pdu;
|
||||||
|
case kineticlib.ops.PUT_RESPONSE:
|
||||||
|
this._events.emit('putResponse', err);
|
||||||
|
break;
|
||||||
|
case kineticlib.ops.DELETE_RESPONSE:
|
||||||
|
this._events.emit('deleteResponse', err);
|
||||||
|
break;
|
||||||
|
case kineticlib.ops.GET_RESPONSE:
|
||||||
|
if (err) {
|
||||||
|
this._events.emit('getResponse', err);
|
||||||
|
} else {
|
||||||
|
this._startGet(socket, pdu);
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
_startGet(socket, pdu) {
|
||||||
|
let chunk = Buffer.allocUnsafe(0);
|
||||||
|
this._request = true;
|
||||||
|
socket._chunkSize = pdu.getChunkSize();
|
||||||
|
if (socket._chunkSize > 0) {
|
||||||
|
chunk = socket.sock.read();
|
||||||
|
this._count += chunk.length;
|
||||||
|
if (this._count === socket._chunkSize) {
|
||||||
|
this._events.emit('getResponse', null, chunk);
|
||||||
|
} else {
|
||||||
|
this._chunkTab.push(chunk);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
this._events.emit('getResponse', errors.ObjNotFound);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
_endGet(socket) {
|
||||||
|
let chunk = Buffer.allocUnsafe(0);
|
||||||
|
if (this._count !== socket._chunkSize) {
|
||||||
|
chunk = socket.sock.read();
|
||||||
|
this._chunkTab.push(chunk);
|
||||||
|
this._count += chunk.length;
|
||||||
|
}
|
||||||
|
if (this._count === socket._chunkSize) {
|
||||||
|
this._events.emit(
|
||||||
|
'getResponse', null, Buffer.concat(this._chunkTab));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
_newSocket() {
|
||||||
|
const socket = new net.Socket({ allowHalfOpen: false }).pause();
|
||||||
|
socket.setKeepAlive(true);
|
||||||
|
socket.unref();
|
||||||
|
return socket;
|
||||||
|
}
|
||||||
|
|
||||||
|
_propError(pdu) {
|
||||||
|
const statusCode = pdu.getStatusCode();
|
||||||
|
// if (statusCode === kineticlib.errors.NOT_FOUND) {
|
||||||
|
// return errors.ObjNotFound;
|
||||||
|
// }
|
||||||
|
if (statusCode !== kineticlib.errors.SUCCESS) {
|
||||||
|
return pdu.getErrorMessage();
|
||||||
|
}
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* set the sequence of the request.
|
||||||
|
*
|
||||||
|
* @param {number} index - the index of the socket.
|
||||||
|
* @param {number} sequence - the sequence to be set.
|
||||||
|
* @returns {Kinetic} - To allow for a functional style.
|
||||||
|
*/
|
||||||
|
setSequence(index, sequence) {
|
||||||
|
if (sequence >= Number.MAX_VALUE) {
|
||||||
|
const temp = this._sockets[index].sock;
|
||||||
|
this._sockets[index].sock = this._newSocket();
|
||||||
|
this._sockets[index].sock.on('connect', () => {
|
||||||
|
this._bindReadable(this._sockets[index], index);
|
||||||
|
});
|
||||||
|
this._sockets[index].sock.connect(this._sockets[index].drive);
|
||||||
|
temp.destroy();
|
||||||
|
this._sockets[index].sequence = 0;
|
||||||
|
} else {
|
||||||
|
this._sockets[index].sequence = sequence;
|
||||||
|
}
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
getSockets() {
|
||||||
|
return this._sockets;
|
||||||
|
}
|
||||||
|
|
||||||
|
_incrString(str) {
|
||||||
|
return (parseInt(str, 36) + 1).toString(36);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* send the put request to the right drive.
|
||||||
|
*
|
||||||
|
* @param {number} index - the index of the socket
|
||||||
|
* @param {Buffer} value - the value to be put.
|
||||||
|
* @param {Object} options - the options object for the put request:
|
||||||
|
* kinetic-protocol
|
||||||
|
* {string} options.synchronization - synchronization mode
|
||||||
|
* {boolean} options.force - option for forcing the put
|
||||||
|
* @param {function} callback - the callback
|
||||||
|
* @returns {undefined}.
|
||||||
|
*/
|
||||||
|
put(index, value, options, callback) {
|
||||||
|
const socket = this._sockets[index];
|
||||||
|
const key = Buffer.concat([Buffer.from(
|
||||||
|
`${socket.drive.host}:${socket.drive.port}:${socket.lexiKey}`)]);
|
||||||
|
socket.lexiKey = this._incrString(socket.lexiKey);
|
||||||
|
const obj = Buffer.concat(value);
|
||||||
|
const tag = crypto.createHmac('sha1', 'asdfasdf').update(obj)
|
||||||
|
.digest();
|
||||||
|
const pdu = new kineticlib.PutPDU(
|
||||||
|
socket.sequence, socket.connectionID, socket.clusterVersion,
|
||||||
|
key, obj.length, tag, options);
|
||||||
|
this.setSequence(socket.index, socket.sequence + 1);
|
||||||
|
const header = pdu.read();
|
||||||
|
const len = header.length + obj.length;
|
||||||
|
return socket.sock.write(Buffer.concat([header, obj], len), err => {
|
||||||
|
if (err) {
|
||||||
|
return callback(err);
|
||||||
|
}
|
||||||
|
return this._events.once(
|
||||||
|
'putResponse', err => callback(err, key));
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* send the get request to the right drive.
|
||||||
|
*
|
||||||
|
* @param {number} index - the index of the socket
|
||||||
|
* @param {Buffer} key - the value to be put.
|
||||||
|
* @param {number[]} range - the range to get
|
||||||
|
* @param {function} callback - the callback
|
||||||
|
* @returns {undefined}.
|
||||||
|
*/
|
||||||
|
get(index, key, range, callback) {
|
||||||
|
const socket = this._sockets[index];
|
||||||
|
const pdu = new kineticlib.GetPDU(
|
||||||
|
socket.sequence, socket.connectionID,
|
||||||
|
socket.clusterVersion, key);
|
||||||
|
this.setSequence(socket.index, socket.sequence + 1);
|
||||||
|
const header = pdu.read();
|
||||||
|
socket.sock.write(header, err => {
|
||||||
|
if (err) {
|
||||||
|
return callback(err);
|
||||||
|
}
|
||||||
|
return this._events.once('getResponse', (err, chunk) => {
|
||||||
|
if (err) {
|
||||||
|
return callback(err);
|
||||||
|
}
|
||||||
|
this._request = false;
|
||||||
|
this._chunkTab = [];
|
||||||
|
this._count = 0;
|
||||||
|
return callback(err, new stream.Readable({
|
||||||
|
read() {
|
||||||
|
this.push(chunk);
|
||||||
|
this.push(null);
|
||||||
|
},
|
||||||
|
}));
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* send the delete request to the right drive.
|
||||||
|
*
|
||||||
|
* @param {number} index - the index of the socket
|
||||||
|
* @param {Buffer} key - the value to be put.
|
||||||
|
* @param {object} options - the options for the delete request
|
||||||
|
* options.force - to force the delete
|
||||||
|
* options.synchronization - the type of the delete
|
||||||
|
* @param {function} callback - the callback
|
||||||
|
* @returns {undefined}.
|
||||||
|
*/
|
||||||
|
delete(index, key, options, callback) {
|
||||||
|
const socket = this._sockets[index];
|
||||||
|
const pdu = new kineticlib.DeletePDU(
|
||||||
|
socket.sequence, socket.connectionID,
|
||||||
|
socket.clusterVersion, key, options);
|
||||||
|
this.setSequence(socket.index, socket.sequence + 1);
|
||||||
|
socket.sock.write(pdu.read(), err => {
|
||||||
|
if (err) {
|
||||||
|
return callback(err);
|
||||||
|
}
|
||||||
|
this._events.removeAllListeners('deleteResponse');
|
||||||
|
return this._events.once(
|
||||||
|
'deleteResponse', err => callback(err));
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export default Kinetic;
|
|
@ -0,0 +1,31 @@
|
||||||
|
import config from '../../Config';
|
||||||
|
|
||||||
|
const backend = {
|
||||||
|
put: function putK(request, size, keyContext, reqUids, callback) {
|
||||||
|
// console.log(util.inspect(config, {showHidden: false, depth: null}));
|
||||||
|
const value = [];
|
||||||
|
const testKinetic = config.kinetic.instance;
|
||||||
|
request.on('data', data => {
|
||||||
|
value.push(data);
|
||||||
|
}).on('end', err => {
|
||||||
|
if (err) {
|
||||||
|
return callback(err);
|
||||||
|
}
|
||||||
|
testKinetic.put(0, value, {}, callback);
|
||||||
|
return undefined;
|
||||||
|
});
|
||||||
|
},
|
||||||
|
|
||||||
|
get: function getK(key, range, reqUids, callback) {
|
||||||
|
const testKinetic = config.kinetic.instance;
|
||||||
|
return testKinetic.get(0, Buffer.from(key), range, callback);
|
||||||
|
},
|
||||||
|
|
||||||
|
delete: function delK(keyValue, reqUids, callback) {
|
||||||
|
const testKinetic = config.kinetic.instance;
|
||||||
|
const key = Buffer.from(keyValue);
|
||||||
|
return testKinetic.delete(0, key, {}, callback);
|
||||||
|
},
|
||||||
|
};
|
||||||
|
|
||||||
|
export default backend;
|
|
@ -1,11 +1,12 @@
|
||||||
import async from 'async';
|
import async from 'async';
|
||||||
import { errors } from 'arsenal';
|
import { errors } from 'arsenal';
|
||||||
|
import assert from 'assert';
|
||||||
import Sproxy from 'sproxydclient';
|
import Sproxy from 'sproxydclient';
|
||||||
import file from './file/backend';
|
import file from './file/backend';
|
||||||
|
import kinetic from './kinetic/backend';
|
||||||
import inMemory from './in_memory/backend';
|
import inMemory from './in_memory/backend';
|
||||||
import config from '../Config';
|
import config from '../Config';
|
||||||
import MD5Sum from '../utilities/MD5Sum';
|
import MD5Sum from '../utilities/MD5Sum';
|
||||||
import assert from 'assert';
|
|
||||||
import kms from '../kms/wrapper';
|
import kms from '../kms/wrapper';
|
||||||
|
|
||||||
let client;
|
let client;
|
||||||
|
@ -17,6 +18,9 @@ if (config.backends.data === 'mem') {
|
||||||
} else if (config.backends.data === 'file') {
|
} else if (config.backends.data === 'file') {
|
||||||
client = file;
|
client = file;
|
||||||
implName = 'file';
|
implName = 'file';
|
||||||
|
} else if (config.backends.data === 'kinetic') {
|
||||||
|
client = kinetic;
|
||||||
|
implName = 'kinetic';
|
||||||
} else if (config.backends.data === 'scality') {
|
} else if (config.backends.data === 'scality') {
|
||||||
client = new Sproxy({
|
client = new Sproxy({
|
||||||
bootstrap: config.sproxyd.bootstrap,
|
bootstrap: config.sproxyd.bootstrap,
|
||||||
|
|
|
@ -28,6 +28,7 @@
|
||||||
"bucketclient": "scality/bucketclient",
|
"bucketclient": "scality/bucketclient",
|
||||||
"commander": "^2.9.0",
|
"commander": "^2.9.0",
|
||||||
"ioctl": "^2.0.0",
|
"ioctl": "^2.0.0",
|
||||||
|
"kinetic": "scality/kineticlib",
|
||||||
"level": "^1.4.0",
|
"level": "^1.4.0",
|
||||||
"level-sublevel": "^6.5.4",
|
"level-sublevel": "^6.5.4",
|
||||||
"multilevel": "^7.3.0",
|
"multilevel": "^7.3.0",
|
||||||
|
|
Loading…
Reference in New Issue