diff --git a/pjs/common/health-check.js b/pjs/common/health-check.js index 32f545c..d840ece 100644 --- a/pjs/common/health-check.js +++ b/pjs/common/health-check.js @@ -1,6 +1,8 @@ (( { config, isDebugEnabled } = pipy.solve('config.js'), + { metrics } = pipy.solve('lib/metrics.js'), + healthCheckTargets = {}, healthCheckServices = {}, @@ -38,7 +40,12 @@ isDebugEnabled && ( console.log('[health-check] ok - service, type, target:', name, type, target.target) ) - ) + ), + metrics.fgwUpstreamStatus.withLabels( + name, + target.ip, + target.port + ).increase() ), fail: target => ( @@ -56,7 +63,12 @@ isDebugEnabled && ( console.log('[health-check] fail - service, type, target:', name, type, target.target) ) - ) + ), + metrics.fgwUpstreamStatus.withLabels( + name, + target.ip, + target.port + ).set(0) ), available: target => ( @@ -113,6 +125,7 @@ healthCheckCache = new algo.Cache(makeHealthCheck), ) => pipy({ + _idx: 0, _service: null, _target: null, _resolve: null, @@ -137,7 +150,10 @@ (_service = healthCheckCache.get(config.Services[name])) && ( Object.keys(config.Services[name].Endpoints || {}).forEach( target => ( + _idx = target.lastIndexOf(':'), healthCheckTargets[target + '@' + name] = { + ip: target.substring(0, _idx), + port: target.substring(_idx + 1), target, service: _service, alive: 1, diff --git a/pjs/http/codec.js b/pjs/http/codec.js index 8d1124e..c70e2d6 100644 --- a/pjs/http/codec.js +++ b/pjs/http/codec.js @@ -1,14 +1,34 @@ -pipy() +(( + { + metrics, + metricsCache, + durationCache, + } = pipy.solve('lib/metrics.js'), +) => pipy() .export('http', { __http: null, }) .pipeline() +.handleStreamStart( + () => ( + metrics.fgwHttpCurrentConnections.withLabels('accepted').increase(), + metrics.fgwHttpCurrentConnections.withLabels('active').increase() + ) +) +.handleStreamEnd( + () => ( + metrics.fgwHttpCurrentConnections.withLabels('handled').increase(), + metrics.fgwHttpCurrentConnections.withLabels('active').decrease() + ) +) .demuxHTTP().to( $=>$ .handleMessageStart( msg => (__http = msg?.head) ) .chain() -) \ No newline at end of file +) + +)() \ No newline at end of file diff --git a/pjs/http/metrics.js b/pjs/http/metrics.js index 97ed935..54ca6fb 100644 --- a/pjs/http/metrics.js +++ b/pjs/http/metrics.js @@ -1,22 +1,38 @@ (( { + metrics, metricsCache, durationCache, } = pipy.solve('lib/metrics.js'), ) => ( pipy({ - _requestTime: null + _request: null, + _requestTime: null, + _requestSize: 0, + _metrics: null, }) .import({ - __service: 'service' + __domain: 'route', + __route: 'route', + __service: 'service', + __target: 'connect-tcp', + __consumer: 'consumer', }) .pipeline() .handleMessageStart( - () => ( - _requestTime = Date.now() + (msg) => ( + _request = msg, + // add HTTP header size + _requestTime = Date.now(), + metrics.fgwHttpRequestsTotal.increase() + ) +) +.handleData( + data => ( + _requestSize += data.size ) ) .chain() @@ -26,21 +42,52 @@ pipy({ serviceName = __service?.name, status = msg?.head?.status, statusClass = Math.floor(status / 100), - metrics = metricsCache.get(serviceName), durationHist = durationCache.get(serviceName), ) => ( durationHist && durationHist.observe(Date.now() - _requestTime), - metrics && ( - metrics.upstreamCompletedCount.increase(), - metrics.upstreamResponseTotal.increase(), + _metrics = metricsCache.get(serviceName), + _metrics && ( + + _metrics.upstreamCompletedCount.increase(), + _metrics.upstreamResponseTotal.increase(), status && ( - metrics.upstreamCodeCount.withLabels(status).increase(), - metrics.upstreamCodeXCount.withLabels(statusClass).increase(), - metrics.upstreamResponseCode.withLabels(statusClass).increase() + + _metrics.fgwHttpStatus.withLabels( + status, + __route?.config?.route || '', + __route?.config?.Path?.Path || '', + __domain?.name || '', + __consumer?.name || '', + __target || '', + (_request?.head?.path || '').split('?')[0] + ).increase(), + + _metrics.fgwBandwidth.withLabels( + 'egress', + __route?.config?.route || '', + __consumer?.name || '', + __inbound.remoteAddress || '' + ).increase(_requestSize), + _requestSize = 0, + // add HTTP header size + + _metrics.upstreamCodeCount.withLabels(status).increase(), + _metrics.upstreamCodeXCount.withLabels(statusClass).increase(), + _metrics.upstreamResponseCode.withLabels(statusClass).increase() ) ) ) )() ) +.handleData( + data => ( + _metrics && _metrics.fgwBandwidth.withLabels( + 'ingress', + __route?.config?.route || '', + __consumer?.name || '', + __inbound.remoteAddress || '' + ).increase(data.size) + ) +) ))() \ No newline at end of file diff --git a/pjs/lib/connect-tcp.js b/pjs/lib/connect-tcp.js index a6da9fd..1d4e025 100644 --- a/pjs/lib/connect-tcp.js +++ b/pjs/lib/connect-tcp.js @@ -12,7 +12,7 @@ pipy({ .export('connect-tcp', { __target: null, __metricLabel: null, - __upstream: {}, + __upstream: null, }) .pipeline() @@ -44,7 +44,7 @@ pipy({ ) .connect(() => __target) .handleStreamEnd( - e => e.error && (__upstream.error = e.error) + e => e.error && (__upstream = {error: e.error}) ) .handleData( data => ( diff --git a/pjs/lib/metrics.js b/pjs/lib/metrics.js index 4b9342f..7960a25 100644 --- a/pjs/lib/metrics.js +++ b/pjs/lib/metrics.js @@ -9,6 +9,24 @@ pod, } = pipy.solve('lib/utils.js'), + fgwHttpStatus = new stats.Counter('fgw_http_status', [ + 'service', 'code', 'route', 'matched_uri', 'matched_host', 'consumer', 'node', 'path' + ]), + + fgwBandwidth = new stats.Counter('fgw_bandwidth', [ + 'service', 'type', 'route', 'consumer', 'node' + ]), + + fgwHttpRequestsTotal = new stats.Gauge('fgw_http_requests_total'), + + fgwHttpCurrentConnections = new stats.Gauge('fgw_http_current_connections', [ + 'state' + ]), + + fgwUpstreamStatus = new stats.Gauge('fgw_upstream_status', [ + 'name', 'ip', 'port' + ]), + sendBytesTotalCounter = new stats.Counter('fgw_service_upstream_cx_tx_bytes_total', [ 'fgw_service_name' ]), @@ -79,8 +97,19 @@ 'fgw_service_name' ]), + metrics = { + fgwHttpRequestsTotal, + fgwHttpCurrentConnections, + fgwUpstreamStatus, + }, + metricsCache = new algo.Cache(serviceName => ( { + fgwHttpStatus: fgwHttpStatus.withLabels(serviceName), + fgwBandwidth: fgwBandwidth.withLabels(serviceName), + fgwHttpRequestsTotal, + fgwHttpCurrentConnections, + sendBytesTotalCounter: sendBytesTotalCounter.withLabels(serviceName), receiveBytesTotalCounter: receiveBytesTotalCounter.withLabels(serviceName), activeConnectionGauge: activeConnectionGauge.withLabels(serviceName), @@ -129,6 +158,7 @@ ), { + metrics, metricsCache, durationCache, rateLimitCounter: new stats.Counter('http_local_rate_limiter', [