Mike Gerwitz

Activist for User Freedom

aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--bin/delta-processor.ts158
-rw-r--r--package.json.in5
-rw-r--r--src/system/AmqpPublisher.ts2
-rw-r--r--src/system/DeltaLogger.ts6
-rw-r--r--src/system/DeltaProcessor.ts54
-rw-r--r--src/system/DeltaPublisher.ts279
-rw-r--r--src/system/MetricsCollector.ts210
-rw-r--r--src/system/db/DeltaDao.ts22
-rw-r--r--src/system/db/MongoDeltaDao.ts91
-rw-r--r--src/types/mongodb.d.ts8
-rw-r--r--test/system/DeltaProcessorTest.ts6
11 files changed, 658 insertions, 183 deletions
diff --git a/bin/delta-processor.ts b/bin/delta-processor.ts
index 836cb83..1dde6df 100644
--- a/bin/delta-processor.ts
+++ b/bin/delta-processor.ts
@@ -18,6 +18,7 @@
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
+import fs = require( 'fs' );
import { AmqpConfig } from "../src/system/AmqpPublisher";
import { MongoDeltaDao } from "../src/system/db/MongoDeltaDao";
@@ -28,27 +29,32 @@ import { DeltaLogger } from "../src/system/DeltaLogger";
import { EventEmitter } from "events";
import { EventDispatcher } from "../src/system/event/EventDispatcher";
import { EventSubscriber } from "../src/system/event/EventSubscriber";
-// import { MetricsCollector } from "../src/system/MetricsCollector";
+import {
+ MetricsCollector,
+ PrometheusConfig,
+} from "../src/system/MetricsCollector";
const {
Db: MongoDb,
Server: MongoServer,
- Connection: MongoConnection,
ReplServers: ReplSetServers,
-} = require( 'mongodb/lib/mongodb' );
+} = require( 'mongodb' );
// TODO: fix this
-process.env.hostname = 'localhost';
-process.env.port = '5672';
-process.env.username = 'quote_referral';
-process.env.password = 'Et7iojahwo4aePie9Cahng7Chu5eim4E';
-process.env.vhost = 'quote';
-process.env.exchange = 'quoteupdate';
-
+process.env.NODE_ENV = 'dev';
+process.env.amqp_hostname = 'localhost';
+process.env.amqp_port = '5672';
+process.env.amqp_username = 'quote_referral';
+process.env.amqp_password = 'Et7iojahwo4aePie9Cahng7Chu5eim4E';
+process.env.amqp_vhost = 'quote';
+process.env.amqp_exchange = 'quoteupdate';
+process.env.prom_hostname = 'dmz2docker01.rsgcorp.local';
+process.env.prom_port = '9091';
// Environment variables
const amqp_conf = _getAmqpConfig( process.env );
const db_conf = _getMongoConfig( process.env );
+const prom_conf = _getPrometheusConfig( process.env );
const env = process.env.NODE_ENV || 'Unknown Environment';
// Event handling
@@ -57,8 +63,8 @@ const event_dispatcher = new EventDispatcher( event_emitter );
const event_subscriber = new EventSubscriber( event_emitter );
// Event subscribers
-new DeltaLogger( env, event_subscriber, ts_ctr ).init();
-// new MetricsCollector( env, event_subscriber );
+new DeltaLogger( env, event_subscriber, ts_ctr );
+const metrics = new MetricsCollector( prom_conf, event_subscriber );
// Instantiate classes for processor
const db = _createDB( db_conf );
@@ -69,11 +75,104 @@ const processor = new DeltaProcessor( dao, publisher, event_dispatcher );
// If the dao intializes successfully then process on a two second interval
const interval_ms = 2000;
+let process_interval: NodeJS.Timer;
+
dao.init()
-.then( _ => { setInterval( () => { processor.process(); }, interval_ms ); } )
+.then( _ =>
+{
+ publisher.connect();
+} )
+.then( _ =>
+{
+ const pidPath = __dirname + '/../conf/.delta_processor.pid';
+
+ writePidFile(pidPath );
+ greet( 'Liza Delta Processor', pidPath );
+
+ process_interval = setInterval(
+ () =>
+ {
+ processor.process();
+ metrics.checkForErrors( dao );
+ },
+ interval_ms,
+ );
+} )
.catch( err => { console.error( 'Mongo Error: ' + err ); } );
+/**
+ * Output greeting
+ *
+ * The greeting contains the program name and PID file path.
+ *
+ * @param name - program name
+ * @param pid_path - path to PID file
+ */
+function greet( name: string, pid_path: string ): void
+{
+ console.log( `${name}`);
+ console.log( `PID file: ${pid_path}` );
+}
+
+
+/**
+ * Write process id (PID) file
+ *
+ * @param pid_path - path to pid file
+ */
+function writePidFile( pid_path: string ): void
+{
+ fs.writeFileSync( pid_path, process.pid );
+
+ process.on( 'SIGINT', function()
+ {
+ shutdown( 'SIGINT' );
+ } )
+ .on( 'SIGTERM', function()
+ {
+ shutdown( 'SIGTERM' );
+ } )
+ .on( 'exit', () =>
+ {
+ fs.unlink( pid_path, () => {} );
+ } );
+}
+
+
+/**
+ * Perform a graceful shutdown
+ *
+ * @param signal - the signal that caused the shutdown
+ */
+function shutdown( signal: string ): void
+{
+ console.log( "Received " + signal + ". Beginning graceful shutdown:" );
+
+ console.log( "...Stopping processing interval" );
+
+ clearInterval( process_interval );
+
+ console.log( "...Closing MongoDb connection" );
+
+ db.close( ( err, _data ) =>
+ {
+ if ( err )
+ {
+ console.error( " Error closing connection: " + err );
+ }
+ } );
+
+ console.log( "...Closing AMQP connection..." );
+
+ publisher.close();
+
+ console.log( "Shutdown complete. Exiting." );
+
+ process.exit();
+}
+
+
/** Timestamp constructor
*
* @return a timestamp
@@ -95,7 +194,7 @@ function _createDB( conf: MongoDbConfig ): MongoDb
{
if( conf.ha )
{
- var mongodbPort = conf.port || MongoConnection.DEFAULT_PORT;
+ var mongodbPort = conf.port || 27017;
var mongodbReplSet = conf.replset || 'rs0';
var dbServers = new ReplSetServers(
[
@@ -109,7 +208,7 @@ function _createDB( conf: MongoDbConfig ): MongoDb
{
var dbServers = new MongoServer(
conf.host || '127.0.0.1',
- conf.port || MongoConnection.DEFAULT_PORT,
+ conf.port || 27017,
{auto_reconnect: true}
);
}
@@ -155,14 +254,31 @@ function _getAmqpConfig( env: any ): AmqpConfig
{
return <AmqpConfig>{
"protocol": "amqp",
- "hostname": env.hostname,
- "port": +( env.port || 0 ),
- "username": env.username,
- "password": env.password,
+ "hostname": env.amqp_hostname,
+ "port": +( env.amqp_port || 0 ),
+ "username": env.amqp_username,
+ "password": env.amqp_password,
"locale": "en_US",
"frameMax": 0,
"heartbeat": 0,
- "vhost": env.vhost,
- "exchange": env.exchange,
+ "vhost": env.amqp_vhost,
+ "exchange": env.amqp_exchange,
+ };
+}
+
+
+/**
+ * Create a prometheus configuration from the environment
+ *
+ * @param env - the environment variables
+ *
+ * @return the prometheus configuration
+ */
+function _getPrometheusConfig( env: any ): PrometheusConfig
+{
+ return <PrometheusConfig>{
+ "hostname": env.prom_hostname,
+ "port": +( env.prom_port || 0 ),
+ "env": process.env.NODE_ENV,
};
} \ No newline at end of file
diff --git a/package.json.in b/package.json.in
index 0aea8e0..898ec54 100644
--- a/package.json.in
+++ b/package.json.in
@@ -30,7 +30,7 @@
},
"devDependencies": {
"typescript": "~3.7",
- "@types/node": "@TS_NODE_VERSION@",
+ "@types/node": "12.12.11",
"chai": ">=1.9.1 < 4",
"@types/chai": ">=1.9.1 < 4",
"chai-as-promised": "7.1.0",
@@ -40,7 +40,8 @@
"sinon": ">=1.17.4",
"es6-promise": "~3",
"@types/amqplib": "0.5.13",
- "avro-js": "1.9.1"
+ "avro-js": "1.9.1",
+ "prom-client": "11.0.0"
},
"licenses": [
diff --git a/src/system/AmqpPublisher.ts b/src/system/AmqpPublisher.ts
index 019ea0d..622d659 100644
--- a/src/system/AmqpPublisher.ts
+++ b/src/system/AmqpPublisher.ts
@@ -38,5 +38,5 @@ export interface AmqpPublisher
*
* @param delta - The delta to publish
*/
- publish( delta: DeltaResult<any> ): Promise<null>;
+ publish( delta: DeltaResult<any> ): Promise<NullableError>;
}
diff --git a/src/system/DeltaLogger.ts b/src/system/DeltaLogger.ts
index 50e616e..f286387 100644
--- a/src/system/DeltaLogger.ts
+++ b/src/system/DeltaLogger.ts
@@ -55,7 +55,9 @@ export class DeltaLogger
private readonly _env: string,
private readonly _subscriber: EventSubscriber,
private readonly _ts_ctr : () => UnixTimestamp,
- ) {}
+ ) {
+ this.init();
+ }
/**
@@ -65,7 +67,7 @@ export class DeltaLogger
{
this._registerEvent( 'document-processed', LogLevel.NOTICE );
this._registerEvent( 'delta-publish', LogLevel.NOTICE );
- this._registerEvent( 'avro-parse-err', LogLevel.ERROR );
+ this._registerEvent( 'avro-err', LogLevel.ERROR );
this._registerEvent( 'mongodb-err', LogLevel.ERROR );
this._registerEvent( 'publish-err', LogLevel.ERROR );
}
diff --git a/src/system/DeltaProcessor.ts b/src/system/DeltaProcessor.ts
index 39cdf3d..db5be5b 100644
--- a/src/system/DeltaProcessor.ts
+++ b/src/system/DeltaProcessor.ts
@@ -59,7 +59,7 @@ export class DeltaProcessor
{
let self = this;
- this._dao.getUnprocessedDocuments()
+ self._dao.getUnprocessedDocuments()
.then( docs =>
{
docs.forEach( doc =>
@@ -68,23 +68,50 @@ export class DeltaProcessor
const doc_id: DocumentId = doc.id;
const last_updated_ts = doc.lastUpdate;
- deltas.forEach( delta =>
+ for ( let i = 0; i < deltas.length; i++ )
{
+ const delta = deltas[ i ];
+ const startTime = process.hrtime();
+ let error = null;
+
self._publisher.publish( delta )
.then( _ =>
{
self._dao.advanceDeltaIndex( doc_id, delta.type );
} )
- .catch( _ =>
+ .catch( err =>
{
- // TODO: blow up?
+ self._dao.setErrorFlag( doc_id );
+
+ error = err;
} );
- });
+
+ // Do not process any more deltas for
+ // this document if there was an error
+ if ( error )
+ {
+ self._dispatcher.dispatch(
+ 'delta-process-error',
+ error
+ );
+
+ return;
+ }
+ else
+ {
+ const elapsedTime = process.hrtime( startTime );
+
+ self._dispatcher.dispatch(
+ 'delta-process-complete',
+ elapsedTime[ 1 ] / 10000
+ );
+ }
+ };
self._dao.markDocumentAsProcessed( doc_id, last_updated_ts )
.then( _ =>
{
- this._dispatcher.dispatch(
+ self._dispatcher.dispatch(
'document-processed',
'Deltas on document ' + doc_id + ' processed '
+ 'successfully. Document has been marked as '
@@ -93,13 +120,13 @@ export class DeltaProcessor
} )
.catch( err =>
{
- this._dispatcher.dispatch( 'mongodb-err', err );
+ self._dispatcher.dispatch( 'mongodb-err', err );
} );
- });
+ } );
} )
.catch( err =>
{
- this._dispatcher.dispatch( 'mongodb-err', err );
+ self._dispatcher.dispatch( 'mongodb-err', err );
} );
}
@@ -137,14 +164,15 @@ export class DeltaProcessor
const deltas: DeltaResult<any>[] = deltas_obj[ type ] || [];
// Get type specific delta index
- let last_published_index = 0;
- if ( doc.lastPublishDelta )
+ let published_count = 0;
+ if ( doc.totalPublishDelta )
{
- last_published_index = doc.lastPublishDelta[ type ] || 0;
+ published_count = doc.totalPublishDelta[ type ] || 0;
}
// Only return the unprocessed deltas
- const deltas_trimmed = deltas.slice( last_published_index );
+ console.log( published_count );
+ const deltas_trimmed = deltas.slice( published_count );
// Mark each delta with its type
deltas_trimmed.forEach( delta =>
diff --git a/src/system/DeltaPublisher.ts b/src/system/DeltaPublisher.ts
index c67cde4..c718756 100644
--- a/src/system/DeltaPublisher.ts
+++ b/src/system/DeltaPublisher.ts
@@ -27,7 +27,8 @@ import { EventDispatcher } from './event/EventDispatcher';
import {
connect as amqpConnect,
Options,
- Channel
+ Channel,
+ Connection,
} from 'amqplib';
const avro = require( 'avro-js' );
@@ -38,8 +39,23 @@ export interface AmqpConfig extends Options.Connect {
}
+export interface AvroSchema {
+ /** Write data to a buffer */
+ toBuffer( data: Record<string, any> ): Buffer | null;
+}
+
+
export class DeltaPublisher implements AmqpPublisher
{
+ /** The amqp connection */
+ private _conn?: Connection;
+
+ /** The amqp channel */
+ private _channel?: Channel;
+
+ /** The avro schema */
+ private _type?: AvroSchema;
+
/** The path to the avro schema */
readonly SCHEMA_PATH = __dirname + '/avro/schema.avsc';
@@ -51,7 +67,7 @@ export class DeltaPublisher implements AmqpPublisher
/**
- * Initialize publisher
+ * Delta publisher
*
* @param _conf - amqp configuration
* @param _emitter - event emitter instance
@@ -61,67 +77,94 @@ export class DeltaPublisher implements AmqpPublisher
private readonly _conf: AmqpConfig,
private readonly _dispatcher: EventDispatcher,
private readonly _ts_ctr : () => UnixTimestamp,
- ) {}
+ ) {
+ this._type = avro.parse( this.SCHEMA_PATH );
+ }
/**
- * Publish quote message to exchange post-rating
- *
- * @param delta - The delta to publish
- *
- * @return whether the message was published successfully
- */
- publish( delta: DeltaResult<any> ): Promise<null>
+ * Initialize connection
+ */
+ connect(): Promise<NullableError>
{
- const exchange = this._conf.exchange;
-
return new Promise<null>( ( resolve, reject ) =>
{
amqpConnect( this._conf )
.then( conn =>
{
- setTimeout( () => conn.close(), 10000 );
- return conn.createChannel();
+ this._conn = conn;
+
+ return this._conn.createChannel();
} )
- .then( ch =>
+ .then( ( ch: Channel ) =>
{
- ch.assertExchange( exchange, 'fanout', { durable: true } );
+ this._channel = ch;
+
+ this._channel.assertExchange(
+ this._conf.exchange,
+ 'fanout',
+ { durable: true }
+ );
- return this.sendMessage( ch, exchange, delta );
+ resolve();
+ return;
} )
- .then( sentSuccessfully =>
+ .catch( e =>
{
- console.log('sentSuccessfully', sentSuccessfully);
- if ( sentSuccessfully )
- {
- this._dispatcher.dispatch(
- 'delta-publish',
- "Published " + delta.type + " delta with ts '"
- + delta.timestamp + "' to '" + exchange
- + '" exchange',
- );
-
- resolve();
- }
- else
- {
- this._dispatcher.dispatch(
- 'publish-err',
- "Error publishing " + delta.type + " delta with ts '"
- + delta.timestamp + "' to '" + exchange
- + "' exchange",
- );
-
- reject();
- }
+ reject( e );
+ return;
+ } );
+ } );
+ }
+
+
+ /**
+ * Close the amqp conenction
+ */
+ close(): void
+ {
+ if ( this._conn )
+ {
+ this._conn.close.bind(this._conn);
+ }
+ }
+
+
+ /**
+ * Publish quote message to exchange post-rating
+ *
+ * @param delta - The delta to publish
+ *
+ * @return whether the message was published successfully
+ */
+ publish( delta: DeltaResult<any> ): Promise<NullableError>
+ {
+ return new Promise<NullableError>( ( resolve, reject ) =>
+ {
+ const startTime = process.hrtime();
+
+ this.sendMessage( delta )
+ .then( _ =>
+ {
+ this._dispatcher.dispatch(
+ 'delta-publish',
+ "Published " + delta.type + " delta with ts '"
+ + delta.timestamp + "' to '" + this._conf.exchange
+ + '" exchange',
+ );
+
+ console.log('#publish: '
+ + process.hrtime( startTime )[0] / 10000 );
+ resolve();
+ return;
} )
.catch( e =>
{
this._dispatcher.dispatch(
'publish-err',
"Error publishing " + delta.type + " delta with ts '"
- + delta.timestamp + '" to "' + exchange + "' exchange '"
- + e,
+ + delta.timestamp + '" to "' + this._conf.exchange
+ + "' exchange: '" + e,
)
reject();
@@ -133,75 +176,92 @@ export class DeltaPublisher implements AmqpPublisher
/**
* Send message to exchange
*
- * @param channel - AMQP channel
- * @param exchange - exchange name
- * @param delta - The delta to publish
+ * @param delta - The delta to publish
*
* @return whether publish was successful
*/
- sendMessage(
- channel: Channel,
- exchange: string,
- delta: DeltaResult<any>,
- ): boolean
+ sendMessage( delta: DeltaResult<any> ): Promise<NullableError>
{
- const headers = {
- version: 1,
- created: Date.now(),
- };
-
- // Convert all delta datums to string for avro
- const delta_data = this.avroFormat( delta.data );
- const event_id = this.DELTA_MAP[ delta.type ];
-
- const data = {
- event: {
- id: event_id,
- ts: this._ts_ctr(),
- actor: 'SERVER',
- step: null,
- },
- document: {
- id: 123123, // Fix
- },
- session: {
- entity_name: 'Foobar', // Fix
- entity_id: 123123, // Fix
- },
- data: {
- Data: {
- bucket: delta_data,
+ return new Promise<NullableError>( ( resolve, reject ) =>
+ {
+ const startTime = process.hrtime();
+
+ const ts = this._ts_ctr();
+ const headers = { version: 1, created: ts };
+ const delta_data = this.avroFormat( delta.data );
+ console.log('#sendmessage 1: '
+ + (process.hrtime( startTime )[ 1 ] / 10000) + 'ms');
+ const event_id = this.DELTA_MAP[ delta.type ];
+ const avro_buffer = this.avroEncode( {
+ event: {
+ id: event_id,
+ ts: ts,
+ actor: 'SERVER',
+ step: null,
},
- },
- delta: {
- Data: {
- bucket: delta_data,
+ document: {
+ id: 123123, // Fix
},
- },
- program: {
- Program: {
- id: 'quote_server',
- version: 'dadaddwafdwa', // Fix
+ session: {
+ entity_name: 'Foobar', // Fix
+ entity_id: 123123, // Fix
},
- },
- };
-
- const avro_buffer = this.avroEncode( data );
+ data: {
+ Data: {
+ bucket: delta_data,
+ },
+ },
+ delta: {
+ Data: {
+ bucket: delta_data,
+ },
+ },
+ program: {
+ Program: {
+ id: 'quote_server',
+ version: 'dadaddwafdwa', // Fix
+ },
+ },
+ } );
+ console.log('#sendmessage 2: '
+ + (process.hrtime( startTime )[ 1 ] / 10000) + 'ms');
- if ( !avro_buffer )
- {
- return false;
- }
+ if ( !this._conn )
+ {
+ reject( 'Error sending message: No connection' );
+ return;
+ }
+ else if ( !this._channel )
+ {
+ reject( 'Error sending message: No channel' );
+ return;
+ }
+ else if ( !avro_buffer )
+ {
+ reject( 'Error sending message: No avro buffer' );
+ return;
+ }
+ console.log('#sendmessage 3: '
+ + (process.hrtime( startTime )[ 1 ] / 10000) + 'ms');
+
+ // we don't use a routing key; fanout exchange
+ const published_successfully = this._channel.publish(
+ this._conf.exchange,
+ '',
+ avro_buffer,
+ { headers: headers },
+ );
- // we don't use a routing key; fanout exchange
- const routing_key = '';
+ if ( published_successfully )
+ {
+ console.log('#sendmessage 4: '
+ + (process.hrtime( startTime )[ 1 ] / 10000) + 'ms');
+ resolve();
+ return;
+ }
- return channel.publish(
- exchange,
- routing_key,
- avro_buffer,
- { headers: headers },
- );
+ reject( 'Error sending message: publishing failed' );
+ } );
}
@@ -218,13 +278,22 @@ export class DeltaPublisher implements AmqpPublisher
try
{
- const type = avro.parse( this.SCHEMA_PATH );
- buffer = type.toBuffer( data );
+ if ( !this._type )
+ {
+ this._dispatcher.dispatch(
+ 'avro-err',
+ 'No avro scheama found',
+ );
+
+ return null;
+ }
+
+ buffer = this._type.toBuffer( data );
}
catch( e )
{
this._dispatcher.dispatch(
- 'avro-parse-err',
+ 'avro-err',
'Error encoding data to avro: ' + e,
);
}
diff --git a/src/system/MetricsCollector.ts b/src/system/MetricsCollector.ts
index 583b5a6..2d7ca24 100644
--- a/src/system/MetricsCollector.ts
+++ b/src/system/MetricsCollector.ts
@@ -22,41 +22,177 @@
*/
import { EventSubscriber } from "./event/EventSubscriber";
+import { DeltaDao } from "./db/DeltaDao";
+import { PositiveInteger } from "../numeric";
+import { Histogram, Pushgateway, Counter, Gauge } from 'prom-client';
-const client = require('prom-client');
+const client = require( 'prom-client' );
-declare type MetricStructure = {
- path: string;
- code: number;
- service: string;
- env: string;
+
+// declare type MetricStructure = {
+// path: string;
+// code: number;
+// service: string;
+// env: string;
+// }
+
+
+export declare type PrometheusConfig = {
+ /** The hostname to connect to */
+ hostname: string;
+
+ /** The port to connect to */
+ port: number;
+
+ /** The environment ( dev, test, demo, live ) */
+ env: string;
}
+
export class MetricsCollector
{
+ /** The prometheus PushGateway */
+ private _gateway: Pushgateway;
+
+ /** Metric push interval */
+ private _push_interval_ms: PositiveInteger = <PositiveInteger>5000;
+
+ /** Delta processed time histogram */
+ private _process_time_hist: Histogram;
+ private _process_time_params: Pushgateway.Parameters = {
+ jobName: 'liza_delta_process_time'
+ };
+
+ /** Delta error counter */
+ private _process_error_count: Counter;
+ private _process_error_params: Pushgateway.Parameters = {
+ jobName: 'liza_delta_error'
+ };
+
+ /** Delta current error gauge */
+ private _current_error_gauge: Gauge;
+ private _current_error_params: Pushgateway.Parameters = {
+ jobName: 'liza_delta_current_error'
+ };
+
+ /** Delta error counter */
+ private _process_delta_count: Counter;
+ private _process_delta_params: Pushgateway.Parameters = {
+ jobName: 'liza_delta_success'
+ };
+
/**
* Initialize delta logger
+ *
+ * @param _conf - the prometheus configuration
+ * @param _subscriber - the event subscriber
*/
constructor(
- private readonly _env: string,
+ private readonly _conf: PrometheusConfig,
private readonly _subscriber: EventSubscriber,
- ) {}
+ ) {
+ // Set labels
+ const default_labels = {
+ env: this._conf.env,
+ service: 'delta_processor',
+ };
+
+ client.register.setDefaultLabels( default_labels );
+
+ // Create gateway
+ const url = 'http://' + this._conf.hostname + ':' + this._conf.port;
+ this._gateway = new client.Pushgateway( url );
+
+ // Create metrics
+ this._process_time_hist = new client.Histogram( {
+ name: this._process_time_params.jobName,
+ help: 'Time in ms for deltas to be processed',
+ labelNames: [ 'env', 'service' ],
+ buckets: client.linearBuckets(0, 10, 10),
+ } );
+
+ this._process_error_count = new client.Counter( {
+ name: this._process_error_params.jobName,
+ help: 'Error count for deltas being processed',
+ labelNames: [ 'env', 'service' ],
+ } );
+
+ this._current_error_gauge = new client.Gauge( {
+ name: this._current_error_params.jobName,
+ help: 'The current number of documents in an error state',
+ labelNames: [ 'env', 'service' ],
+ } );
+
+ this._process_delta_count = new client.Counter( {
+ name: this._process_delta_params.jobName,
+ help: 'Count of deltas successfully processed',
+ labelNames: [ 'env', 'service' ],
+ } );
+
+ // Push metrics on a specific intervals
+ setInterval( () => { this.pushMetrics(); }, this._push_interval_ms );
+
+ // Subsribe metrics to events
+ this.subscribeMetrics();
+ }
/**
- * Initialize the logger to look for specific events
+ * Subscribe metrics
*/
- init(): void
+ private subscribeMetrics()
{
- const collectDefaultMetrics = client.collectDefaultMetrics;
-
- console.log( this._subscriber, collectDefaultMetrics)
- this._formatLog( '', 123 );
- // this._registerEvent( 'document-processed', LogLevel.NOTICE );
- // this._registerEvent( 'delta-publish', LogLevel.NOTICE );
- // this._registerEvent( 'avro-parse-err', LogLevel.ERROR );
- // this._registerEvent( 'mongodb-err', LogLevel.ERROR );
- // this._registerEvent( 'publish-err', LogLevel.ERROR );
+ this._subscriber.subscribe(
+ 'delta-process-complete',
+ ( val ) =>
+ {
+ console.log( 'Got time: ' + val + 'ms' );
+ this._process_time_hist.observe( val );
+ this._process_delta_count.inc();
+ }
+ );
+
+ this._subscriber.subscribe(
+ 'delta-process-error',
+ ( _ ) => this._process_error_count.inc()
+ );
+ }
+
+
+ /**
+ * Push metrics to Prometheus PushGateway
+ */
+ private pushMetrics(): void
+ {
+ // this._gateway.pushAdd( this._process_time_params, this.pushCallback );
+ // this._gateway.pushAdd( this._process_error_params, this.pushCallback );
+ // this._gateway.pushAdd( this._current_error_params, this.pushCallback );
+ // this._gateway.pushAdd( this._process_delta_params, this.pushCallback );
+
+ this._gateway.pushAdd(
+ {
+ jobName: 'liza_delta_metrics'
+ },
+ this.pushCallback
+ );
+ }
+
+
+ /**
+ * Handle push error
+ *
+ * @param error - Any errors that occurred
+ * @param response - The http response
+ * @param body - The resposne body
+ */
+ private pushCallback(
+ _error?: Error | undefined,
+ _response?: any,
+ _body?: any
+ ): void
+ {
+ // console.log( 'Push callback' );
+ // console.error( error, response, body );
}
@@ -68,13 +204,35 @@ export class MetricsCollector
*
* @returns a structured logging object
*/
- private _formatLog( path: string, code: number ): MetricStructure
+ // private _formatMetricVal( label: string, val: any ): MetricStructure
+ // {
+ // return <MetricStructure>{
+ // path: path,
+ // code: code,
+ // service: 'quote-server',
+ // env: this._conf.env,
+ // };
+ // }
+
+
+ /**
+ * Look for mongodb delta errors and update metrics if found
+ *
+ * @return any errors the occurred
+ */
+ checkForErrors( dao: DeltaDao ): NullableError
{
- return <MetricStructure>{
- path: path,
- code: code,
- service: 'quote-server',
- env: this._env,
- };
+ dao.getErrorCount()
+ .then( count =>
+ {
+ // console.log( 'Error count: ', count );
+ this._current_error_gauge.set( +count );
+ } )
+ .catch( err =>
+ {
+ return err;
+ } );
+
+ return null;
}
}
diff --git a/src/system/db/DeltaDao.ts b/src/system/db/DeltaDao.ts
index 64e1f0f..173fcee 100644
--- a/src/system/db/DeltaDao.ts
+++ b/src/system/db/DeltaDao.ts
@@ -47,7 +47,7 @@ export interface DeltaDao
* @param doc_id - Document whose index will be set
* @param type - Delta type
*
- * @return any errors that occured
+ * @return any errors that occurred
*/
advanceDeltaIndex(
doc_id: DocumentId,
@@ -62,11 +62,29 @@ export interface DeltaDao
* @param doc_id - The document to mark
* @param last_update_ts - The last time this document was updated
*
- * @return true if the document was successfully marked as processed
+ * @return any errors that occurred
*/
markDocumentAsProcessed(
doc_id: DocumentId,
last_update_ts: UnixTimestamp,
): Promise<NullableError>
+
+
+ /**
+ * Flag the document as being in an error state
+ *
+ * @param doc_id - The document to flag
+ *
+ * @return any errors that occurred
+ */
+ setErrorFlag( doc_id: DocumentId ): Promise<NullableError>
+
+
+ /**
+ * Get a count of documents in an error state
+ *
+ * @return a count of the documents in an error state
+ */
+ getErrorCount(): Promise<number | Error>
}
diff --git a/src/system/db/MongoDeltaDao.ts b/src/system/db/MongoDeltaDao.ts
index 443ae85..81816e0 100644
--- a/src/system/db/MongoDeltaDao.ts
+++ b/src/system/db/MongoDeltaDao.ts
@@ -59,7 +59,7 @@ export class MongoDeltaDao implements DeltaDao
*
* connectError event will be emitted on failure.
*
- * @return any errors that occured
+ * @return any errors that occurred
*/
init(): Promise<NullableError>
{
@@ -73,14 +73,18 @@ export class MongoDeltaDao implements DeltaDao
// if there was an error, don't bother with anything else
if ( err )
{
- // in some circumstances, it may just be telling us that we're
- // already connected (even though the connection may have been
- // broken)
+ // in some circumstances, it may just be telling us that
+ // we're already connected (even though the connection may
+ // have been broken)
if ( err.errno !== undefined )
{
reject( 'Error opening mongo connection: ' + err );
return;
}
+ } else if ( db == null )
+ {
+ reject( 'No database connection' );
+ return;
}
// quotes collection
@@ -116,7 +120,7 @@ export class MongoDeltaDao implements DeltaDao
);
}
);
- });
+ } );
} );
}
@@ -165,8 +169,10 @@ export class MongoDeltaDao implements DeltaDao
/**
* Set the document's processed index
*
- * @param doc_id - Document whose index will be set
- * @param type - Delta type
+ * @param doc_id - Document whose index will be set
+ * @param type - Delta type
+ *
+ * @return any errors that occurred
*/
advanceDeltaIndex(
doc_id: DocumentId,
@@ -177,7 +183,7 @@ export class MongoDeltaDao implements DeltaDao
{
const inc_data: Record<string, any> = {};
- inc_data[ 'lastPublishDelta.' + type ] = 1;
+ inc_data[ 'totalPublishDelta.' + type ] = 1;
this._collection!.update(
{ id: doc_id },
@@ -206,7 +212,7 @@ export class MongoDeltaDao implements DeltaDao
* @param doc_id - The document to mark
* @param last_update_ts - The last time this document was updated
*
- * @return true if the document was successfully marked as processed
+ * @return any errors that occurred
*/
markDocumentAsProcessed(
doc_id: DocumentId,
@@ -231,7 +237,74 @@ export class MongoDeltaDao implements DeltaDao
return;
}
);
+ } );
+ }
+
+ /**
+ * Flag the document as being in an error state
+ *
+ * @param doc_id - The document to flag
+ *
+ * @return any errors that occurred
+ */
+ setErrorFlag( doc_id: DocumentId ): Promise<NullableError>
+ {
+ return new Promise( ( resolve, reject ) =>
+ {
+ this._collection!.update(
+ { id: doc_id },
+ { $set: { deltaError: true } },
+ { upsert: false },
+ function( err )
+ {
+ if ( err )
+ {
+ reject( "Failed setting error flag: " + err );
+ return;
+ }
+
+ resolve();
+ return;
+ }
+ );
+ } );
+ }
+
+
+ /**
+ * Get a count of documents in an error state
+ *
+ * @return a count of the documents in an error state
+ */
+ getErrorCount(): Promise<number | Error>
+ {
+ return new Promise( ( resolve, reject ) =>
+ {
+ this._collection!.find(
+ { deltaError: true },
+ {},
+ function( err, cursor )
+ {
+ if ( err )
+ {
+ reject( err );
+ return;
+ }
+
+ cursor.toArray( function( err: NullableError, data: any[] )
+ {
+ if ( err )
+ {
+ reject( err );
+ return;
+ }
+
+ // return the count
+ resolve( data.length );
+ });
+ }
+ )
} );
}
}
diff --git a/src/types/mongodb.d.ts b/src/types/mongodb.d.ts
index 61d764e..968fa67 100644
--- a/src/types/mongodb.d.ts
+++ b/src/types/mongodb.d.ts
@@ -51,6 +51,14 @@ export interface MongoDb
* @param callback continuation on completion
*/
open( callback: MongoCallback ): void;
+
+
+ /**
+ * Close the database connection
+ *
+ * @param callback continuation on completion
+ */
+ close( callback: MongoCallback ): void;
}
diff --git a/test/system/DeltaProcessorTest.ts b/test/system/DeltaProcessorTest.ts
index d32197e..bccd4a4 100644
--- a/test/system/DeltaProcessorTest.ts
+++ b/test/system/DeltaProcessorTest.ts
@@ -241,7 +241,7 @@ describe( 'system.DeltaProcessor', () =>
},
],
},
- lastPublishDelta: {
+ totalPublishDelta: {
data: 0,
},
},
@@ -274,7 +274,7 @@ describe( 'system.DeltaProcessor', () =>
},
],
},
- lastPublishDelta: {
+ totalPublishDelta: {
data: 1,
},
},
@@ -308,6 +308,8 @@ function createMockDeltaDao(): DeltaDao
getUnprocessedDocuments() { return Promise.resolve( [] ); },
advanceDeltaIndex() { return Promise.resolve( null ); },
markDocumentAsProcessed() { return Promise.resolve( null ); },
+ setErrorFlag() { return Promise.resolve( null ); },
+ getErrorCount() { return Promise.resolve( 0 ); },
};
}