WIP - remove the balancing between drives

TODO - use index instead of socket

config.json

@@ -27,6 +27,7 @@
         "host": "localhost",
         "port": 8500
     },
+    "kinetic": ["localhost:8123"],
     "clusters": 10,
     "log": {
         "logLevel": "info",

lib/Config.js

@@ -1,7 +1,7 @@
 import assert from 'assert';
 import fs from 'fs';
 import path from 'path';
-
+import Kinetic from './data/kinetic/Kinetic';
 import authDataChecker from './auth/in_memory/checker';
 
 /**
@@ -279,6 +279,20 @@ class Config {
             dataPath,
             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;
     }
 }

lib/data/kinetic/Kinetic.js

@@ -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;

lib/data/kinetic/backend.js

@@ -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;

lib/data/wrapper.js

@@ -1,11 +1,12 @@
 import async from 'async';
 import { errors } from 'arsenal';
+import assert from 'assert';
 import Sproxy from 'sproxydclient';
 import file from './file/backend';
+import kinetic from './kinetic/backend';
 import inMemory from './in_memory/backend';
 import config from '../Config';
 import MD5Sum from '../utilities/MD5Sum';
-import assert from 'assert';
 import kms from '../kms/wrapper';
 
 let client;
@@ -17,6 +18,9 @@ if (config.backends.data === 'mem') {
 } else if (config.backends.data === 'file') {
     client = file;
     implName = 'file';
+} else if (config.backends.data === 'kinetic') {
+    client = kinetic;
+    implName = 'kinetic';
 } else if (config.backends.data === 'scality') {
     client = new Sproxy({
         bootstrap: config.sproxyd.bootstrap,

package.json

@@ -28,6 +28,7 @@
     "bucketclient": "scality/bucketclient",
     "commander": "^2.9.0",
     "ioctl": "^2.0.0",
+    "kinetic": "scality/kineticlib",
     "level": "^1.4.0",
     "level-sublevel": "^6.5.4",
     "multilevel": "^7.3.0",