import { Clock } from '../infrastructure/clock' ; import * as follow_redirects from '../infrastructure/follow_redirects' ; import { JsonConfig } from '../infrastructure/json_config' ; import * as logging from '../infrastructure/logging' ; import { PrometheusClient } from '../infrastructure/prometheus_scraper' ; import { AccessKeyId , AccessKeyMetricsId } from '../model/access_key' ; import { version } from '../package.json' ; import { AccessKeyConfigJson } from './server_access_key' ; import { ServerConfigJson } from './server_config' ; const MS_PER_HOUR = 60 * 60 * 1000 ; const MS_PER_DAY = 24 * MS_PER_HOUR ; const SANCTIONED_COUNTRIES = new Set ( [ 'CU' , 'KP' , 'SY' ] ) ; export interface KeyUsage { accessKeyId : string ; countries : string [ ] ; inboundBytes : number ; } export interface HourlyServerMetricsReportJson { serverId : string ; startUtcMs : number ; endUtcMs : number ; userReports : HourlyUserMetricsReportJson [ ] ; } export interface HourlyUserMetricsReportJson { userId : string ; countries : string [ ] ; bytesTransferred : number ; } export interface DailyFeatureMetricsReportJson { serverId : string ; serverVersion : string ; timestampUtcMs : number ; dataLimit : DailyDataLimitMetricsReportJson ; } export interface DailyDataLimitMetricsReportJson { enabled : boolean ; perKeyLimitCount ? : number ; } export interface SharedMetricsPublisher { startSharing ( ) ; stopSharing ( ) ; isSharingEnabled ( ) ; } export interface UsageMetrics { getUsage ( ) : Promise < KeyUsage [ ] > ; reset ( ) ; } export class PrometheusUsageMetrics implements UsageMetrics { private resetTimeMs : number = Date . now ( ) ; constructor ( private prometheusClient ) { } async getUsage ( ) { const timeDeltaSecs = Math . round ( ( Date . now ( ) - this . resetTimeMs ) / 1000 ) ; const result = await this . prometheusClient . query ( `sum(increase(shadowsocks_data_bytes{dir=~"p>t|p<t"}[${ timeDeltaSecs }s])) by (location, access_key)` ) ; const usage = [ ] as KeyUsage [ ] ; for ( const entry of result . result ) { const accessKeyId = entry . metric [ 'access_key' ] || '' ; let countries = [ ] ; const countriesStr = entry . metric [ 'location' ] || '' ; if ( countriesStr ) { countries = countriesStr . split ( ',' ) . map ( ( e ) => e . trim ( ) ) ; } const inboundBytes = Math . round ( parseFloat ( entry . value [ 1 ] ) ) ; usage . push ( { accessKeyId , countries , inboundBytes } ) ; } return usage ; } reset ( ) { this . resetTimeMs = Date . now ( ) ; } } export interface MetricsCollectorClient { collectServerUsageMetrics ( reportJson ) : Promise < void > ; collectFeatureMetrics ( reportJson ) : Promise < void > ; } export class RestMetricsCollectorClient { constructor ( private serviceUrl ) { } collectServerUsageMetrics ( reportJson ) { return this . postMetrics ( '/connections' , JSON . stringify ( reportJson ) ) ; } collectFeatureMetrics ( reportJson ) { return this . postMetrics ( '/features' , JSON . stringify ( reportJson ) ) ; } private async postMetrics ( urlPath , reportJson ) { const options = { headers : { 'Content-Type' : 'application/json' } , method : 'POST' , body : reportJson } ; const url = `${ this . serviceUrl }${ urlPath }` ; logging . info ( `Posting metrics to ${ url } with options ${ JSON . stringify ( options ) }` ) ; try { const response = await follow_redirects . requestFollowRedirectsWithSameMethodAndBody ( url , options ) ; if ( ! response . ok ) { throw new Error ( `Got status ${ response . status }` ) ; } } catch ( e ) { throw new Error ( `Failed to post to metrics server: ${ e }` ) ; } } } export class OutlineSharedMetricsPublisher implements SharedMetricsPublisher { private reportStartTimestampMs : number ; constructor ( private clock , private serverConfig , private keyConfig , usageMetrics , private toMetricsId , private metricsCollector ) { this . reportStartTimestampMs = this . clock . now ( ) ; this . clock . setInterval ( async ( ) => { if ( ! this . isSharingEnabled ( ) ) { return ; } try { await this . reportServerUsageMetrics ( await usageMetrics . getUsage ( ) ) ; usageMetrics . reset ( ) ; } catch ( err ) { logging . error ( `Failed to report server usage metrics: ${ err }` ) ; } } , MS_PER_HOUR ) ; this . clock . setInterval ( async ( ) => { if ( ! this . isSharingEnabled ( ) ) { return ; } try { await this . reportFeatureMetrics ( ) ; } catch ( err ) { logging . error ( `Failed to report feature metrics: ${ err }` ) ; } } , MS_PER_DAY ) ; } startSharing ( ) { this . serverConfig . data ( ) . metricsEnabled = true ; this . serverConfig . write ( ) ; } stopSharing ( ) { this . serverConfig . data ( ) . metricsEnabled = false ; this . serverConfig . write ( ) ; } isSharingEnabled ( ) { return this . serverConfig . data ( ) . metricsEnabled || false ; } private async reportServerUsageMetrics ( usageMetrics ) { const reportEndTimestampMs = this . clock . now ( ) ; const userReports = [ ] as HourlyUserMetricsReportJson [ ] ; for ( const keyUsage of usageMetrics ) { if ( keyUsage . inboundBytes === 0 ) { continue ; } if ( hasSanctionedCountry ( keyUsage . countries ) ) { continue ; } userReports . push ( { userId : this . toMetricsId ( keyUsage . accessKeyId ) || '' , bytesTransferred : keyUsage . inboundBytes , countries : [ ... keyUsage . countries ] } ) ; } const report = { serverId : this . serverConfig . data ( ) . serverId , startUtcMs : this . reportStartTimestampMs , endUtcMs : reportEndTimestampMs , userReports } as HourlyServerMetricsReportJson ; this . reportStartTimestampMs = reportEndTimestampMs ; if ( userReports . length === 0 ) { return ; } await this . metricsCollector . collectServerUsageMetrics ( report ) ; } private async reportFeatureMetrics ( ) { const keys = this . keyConfig . data ( ) . accessKeys ; const featureMetricsReport = { serverId : this . serverConfig . data ( ) . serverId , serverVersion : version , timestampUtcMs : this . clock . now ( ) , dataLimit : { enabled : ! ! this . serverConfig . data ( ) . accessKeyDataLimit , perKeyLimitCount : keys . filter ( key => ! ! key . dataLimit ) . length } } ; await this . metricsCollector . collectFeatureMetrics ( featureMetricsReport ) ; } } function hasSanctionedCountry ( countries ) { for ( const country of countries ) { if ( SANCTIONED_COUNTRIES . has ( country ) ) { return true ; } } return false ; }