@ -1,24 +1,36 @@
// Libraries
import { isEmpty , isString , fromPairs , map as lodashMap } from 'lodash' ;
import { isEmpty , map as lodashMap , fromPairs } from 'lodash' ;
import { Observable , from , merge , of , iif , defer } from 'rxjs' ;
import { map , filter , catchError , switchMap , mergeMap } from 'rxjs/operators' ;
// Services & Utils
import { dateMath } from '@grafana/data' ;
import { addLabelToSelector } from 'app/plugins/datasource/prometheus/add_label_to_query' ;
import { BackendSrv , DatasourceRequestOptions } from 'app/core/services/backend_srv' ;
import { TemplateSrv } from 'app/features/templating/template_srv' ;
import { safeStringifyValue , convertToWebSocketUrl } from 'app/core/utils/explore' ;
import {
lokiResultsToTableModel ,
processRangeQueryResponse ,
legacyLogStreamToDataFrame ,
lokiStreamResultToDataFrame ,
isLokiLogsStream ,
} from './result_transformer' ;
import { formatQuery , parseQuery , getHighlighterExpressionsFromQuery } from './query_utils' ;
// Types
import {
dateMath ,
DataFrame ,
LogRowModel ,
DateTime ,
LoadingState ,
AnnotationEvent ,
DataFrameView ,
LoadingState ,
TimeRange ,
FieldConfig ,
ArrayVector ,
FieldType ,
FieldConfig ,
} from '@grafana/data' ;
import { addLabelToSelector } from 'app/plugins/datasource/prometheus/add_label_to_query' ;
import LanguageProvider from './language_provider' ;
import { logStreamToDataFrame } from './result_transformer' ;
import { formatQuery , parseQuery , getHighlighterExpressionsFromQuery } from './query_utils' ;
// Types
import {
DataFrame ,
TimeSeries ,
PluginMeta ,
DataSourceApi ,
DataSourceInstanceSettings ,
@ -27,29 +39,38 @@ import {
DataQueryResponse ,
AnnotationQueryRequest ,
} from '@grafana/data' ;
import { LokiQuery , LokiOptions , LokiLogsStream , LokiResponse } from './types' ;
import { BackendSrv } from 'app/core/services/backend_srv' ;
import { TemplateSrv } from 'app/features/templating/template_srv' ;
import { safeStringifyValue , convertToWebSocketUrl } from 'app/core/utils/explore' ;
import { LiveTarget , LiveStreams } from './live_streams' ;
import { Observable , from , merge , of } from 'rxjs' ;
import { map , filter } from 'rxjs/operators' ;
import {
LokiQuery ,
LokiOptions ,
LokiLegacyQueryRequest ,
LokiLegacyStreamResponse ,
LokiResponse ,
LokiResultType ,
LokiRangeQueryRequest ,
LokiStreamResponse ,
LokiLegacyStreamResult ,
} from './types' ;
import { ExploreMode } from 'app/types' ;
import { LegacyTarget , LiveStreams } from './live_streams' ;
import LanguageProvider from './language_provider' ;
type RangeQueryOptions = Pick < DataQueryRequest < LokiQuery > , 'range' | 'intervalMs' | 'maxDataPoints' | 'reverse' > ;
export const DEFAULT_MAX_LINES = 1000 ;
const LEGACY_QUERY_ENDPOINT = '/api/prom/query' ;
const RANGE_QUERY_ENDPOINT = '/loki/api/v1/query_range' ;
const INSTANT_QUERY_ENDPOINT = '/loki/api/v1/query' ;
const DEFAULT_QUERY_PARAMS = {
const DEFAULT_QUERY_PARAMS : Partial < LokiLegacyQueryRequest > = {
direction : 'BACKWARD' ,
limit : DEFAULT_MAX_LINES ,
regexp : '' ,
query : '' ,
} ;
function serializeParams ( data : any ) {
function serializeParams ( data : Record < string , any > ) {
return Object . keys ( data )
. map ( k = > {
const v = data [ k ] ;
return encodeURIComponent ( k ) + '=' + encodeURIComponent ( v ) ;
} )
. map ( k = > ` ${ encodeURIComponent ( k ) } = ${ encodeURIComponent ( data [ k ] ) } ` )
. join ( '&' ) ;
}
@ -62,6 +83,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
private streams = new LiveStreams ( ) ;
languageProvider : LanguageProvider ;
maxLines : number ;
version : string ;
/** @ngInject */
constructor (
@ -70,179 +92,296 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
private templateSrv : TemplateSrv
) {
super ( instanceSettings ) ;
this . languageProvider = new LanguageProvider ( this ) ;
const settingsData = instanceSettings . jsonData || { } ;
this . maxLines = parseInt ( settingsData . maxLines , 10 ) || DEFAULT_MAX_LINES ;
}
_request ( apiUrl : string , data? : any , options? : any ) {
getVersion() {
if ( this . version ) {
return Promise . resolve ( this . version ) ;
}
return this . _request ( RANGE_QUERY_ENDPOINT )
. toPromise ( )
. then ( ( ) = > {
this . version = 'v1' ;
return this . version ;
} )
. catch ( ( err : any ) = > {
this . version = err . status !== 404 ? 'v1' : 'v0' ;
return this . version ;
} ) ;
}
_request ( apiUrl : string , data? : any , options? : DatasourceRequestOptions ) : Observable < Record < string , any > > {
const baseUrl = this . instanceSettings . url ;
const params = data ? serializeParams ( data ) : '' ;
const url = ` ${ baseUrl } ${ apiUrl } ? ${ params } ` ;
const url = ` ${ baseUrl } ${ apiUrl } ${ params . length ? ` ? ${ params } ` : '' } ` ;
const req = {
. . . options ,
url ,
} ;
return this . backendSrv . datasourceRequest ( req ) ;
return from ( this . backendSrv . datasourceRequest ( req ) ) ;
}
prepareLiveTarget ( target : LokiQuery , options : DataQueryRequest < LokiQuery > ) : LiveTarget {
const interpolated = this . templateSrv . replace ( target . expr , { } , this . interpolateQueryExpr ) ;
const { query , regexp } = parseQuery ( interpolated ) ;
const refId = target . refId ;
const baseUrl = this . instanceSettings . url ;
const params = serializeParams ( { query , regexp } ) ;
const url = convertToWebSocketUrl ( ` ${ baseUrl } /api/prom/tail? ${ params } ` ) ;
query ( options : DataQueryRequest < LokiQuery > ) : Observable < DataQueryResponse > {
const subQueries : Array < Observable < DataQueryResponse > > = [ ] ;
const filteredTargets = options . targets
. filter ( target = > target . expr && ! target . hide )
. map ( target = > ( {
. . . target ,
expr : this.templateSrv.replace ( target . expr , { } , this . interpolateQueryExpr ) ,
} ) ) ;
if ( options . exploreMode === ExploreMode . Metrics ) {
filteredTargets . forEach ( target = >
subQueries . push (
this . runInstantQuery ( target , options , filteredTargets . length ) ,
this . runRangeQueryWithFallback ( target , options , filteredTargets . length )
)
) ;
} else {
filteredTargets . forEach ( target = >
subQueries . push (
this . runRangeQueryWithFallback ( target , options , filteredTargets . length ) . pipe (
map ( dataQueryResponse = > {
if ( options . exploreMode === ExploreMode . Logs && dataQueryResponse . data . find ( d = > isTimeSeries ( d ) ) ) {
throw new Error (
'Logs mode does not support queries that return time series data. Please perform a logs query or switch to Metrics mode.'
) ;
} else {
return dataQueryResponse ;
}
} )
)
)
) ;
}
return {
query ,
regexp ,
url ,
refId ,
size : Math.min ( options . maxDataPoints || Infinity , this . maxLines ) ,
} ;
// No valid targets, return the empty result to save a round trip.
if ( isEmpty ( subQueries ) ) {
return of ( {
data : [ ] ,
state : LoadingState.Done ,
} ) ;
}
return merge ( . . . subQueries ) ;
}
prepareQueryTarget ( target : LokiQuery , options : DataQueryRequest < LokiQuery > ) {
const interpolated = this . templateSrv . replace ( target . expr , { } , this . interpolateQueryExpr ) ;
const { query , regexp } = parseQuery ( interpolated ) ;
const start = this . getTime ( options . range . from , false ) ;
const end = this . getTime ( options . range . to , true ) ;
const refId = target . refId ;
return {
runLegacyQuery = (
target : LokiQuery ,
options : { range? : TimeRange ; maxDataPoints? : number ; reverse? : boolean }
) : Observable < DataQueryResponse > = > {
if ( target . liveStreaming ) {
return this . runLiveQuery ( target , options ) ;
}
const range = options . range
? { start : this.getTime ( options . range . from , false ) , end : this.getTime ( options . range . to , true ) }
: { } ;
const query : LokiLegacyQueryRequest = {
. . . DEFAULT_QUERY_PARAMS ,
query ,
regexp ,
start ,
end ,
. . . parseQuery ( target . expr ) ,
. . . range ,
limit : Math.min ( options . maxDataPoints || Infinity , this . maxLines ) ,
refId ,
} ;
}
processError = ( err : any , target : any ) : DataQueryError = > {
const error : DataQueryError = {
message : ( err && err . statusText ) || 'Unknown error during query transaction. Please check JS console logs.' ,
refId : target.refId ,
} ;
if ( err . data ) {
if ( typeof err . data === 'string' ) {
error . message = err . data ;
} else if ( err . data . error ) {
error . message = safeStringifyValue ( err . data . error ) ;
}
} else if ( err . message ) {
error . message = err . message ;
} else if ( typeof err === 'string' ) {
error . message = err ;
}
error . status = err . status ;
error . statusText = err . statusText ;
return error ;
return this . _request ( LEGACY_QUERY_ENDPOINT , query ) . pipe (
catchError ( ( err : any ) = > this . throwUnless ( err , err . cancelled , target ) ) ,
filter ( ( response : any ) = > ! response . cancelled ) ,
map ( ( response : { data : LokiLegacyStreamResponse } ) = > ( {
data : this.lokiLegacyStreamsToDataframes ( response . data , query , this . maxLines , options . reverse ) ,
key : ` ${ target . refId } _log ` ,
} ) )
) ;
} ;
processResult = ( data : LokiLogsStream | LokiResponse , target : any ) : DataFrame [ ] = > {
const series : DataFrame [ ] = [ ] ;
lokiLegacyStreamsToDataframes = (
data : LokiLegacyStreamResult | LokiLegacyStreamResponse ,
target : { refId : string ; query? : string ; regexp? : string } ,
limit : number ,
reverse = false
) : DataFrame [ ] = > {
if ( Object . keys ( data ) . length === 0 ) {
return series ;
return [ ] ;
}
if ( ! ( data as any ) . streams ) {
return [ logStreamToDataFrame ( data as LokiLogsStream , false , target . refId ) ] ;
if ( isLokiLogsStream ( data ) ) {
return [ legacyL ogStreamToDataFrame ( data , false , target . refId ) ] ;
}
data = data as LokiResponse ;
for ( const stream of data . streams || [ ] ) {
const dataFrame = logStreamToDataFrame ( stream ) ;
const series : DataFrame [ ] = data . streams . map ( stream = > {
const dataFrame = legacyLogStreamToDataFrame ( stream , reverse ) ;
this . enhanceDataFrame ( dataFrame ) ;
dataFrame . refId = target . refId ;
dataFrame . meta = {
searchWords : getHighlighterExpressionsFromQuery ( formatQuery ( target . query , target . regexp ) ) ,
limit : this.maxLines ,
return {
. . . dataFrame ,
refId : target.refId ,
meta : {
searchWords : getHighlighterExpressionsFromQuery ( formatQuery ( target . query , target . regexp ) ) ,
limit : this.maxLines ,
} ,
} ;
series . push ( dataFrame ) ;
}
} ) ;
return series ;
} ;
/ * *
* Runs live queries which in this case means creating a websocket and listening on it for new logs .
* This returns a bit different dataFrame than runQueries as it returns single dataframe even if there are multiple
* Loki streams , sets only common labels on dataframe . labels and has additional dataframe . fields . labels for unique
* labels per row .
* /
runLiveQuery = ( options : DataQueryRequest < LokiQuery > , target : LokiQuery ) : Observable < DataQueryResponse > = > {
const liveTarget = this . prepareLiveTarget ( target , options ) ;
const stream = this . streams . getStream ( liveTarget ) ;
return stream . pipe (
map ( data = > {
runInstantQuery = (
target : LokiQuery ,
options : DataQueryRequest < LokiQuery > ,
responseListLength : number
) : Observable < DataQueryResponse > = > {
const timeNs = this . getTime ( options . range . to , true ) ;
const query = {
query : parseQuery ( target . expr ) . query ,
time : ` ${ timeNs + ( 1 e9 - ( timeNs % 1 e9 ) ) } ` ,
limit : Math.min ( options . maxDataPoints || Infinity , this . maxLines ) ,
} ;
return this . _request ( INSTANT_QUERY_ENDPOINT , query ) . pipe (
catchError ( ( err : any ) = > this . throwUnless ( err , err . cancelled , target ) ) ,
filter ( ( response : any ) = > ( response . cancelled ? false : true ) ) ,
map ( ( response : { data : LokiResponse } ) = > {
if ( response . data . data . resultType === LokiResultType . Stream ) {
throw new Error ( 'Metrics mode does not support logs. Use an aggregation or switch to Logs mode.' ) ;
}
return {
data ,
key : ` loki- ${ liveTarget . refId } ` ,
state : LoadingState.Streaming ,
data : [ lokiResultsToTableModel ( response . data . data . result , responseListLength , target . refId , true ) ] ,
key : ` ${ target . refId } _instant ` ,
} ;
} )
) ;
} ;
runQuery = ( options : DataQueryRequest < LokiQuery > , target : LokiQuery ) : Observable < DataQueryResponse > = > {
const query = this . prepareQueryTarget ( target , options ) ;
return from (
this . _request ( '/api/prom/query' , query ) . catch ( ( err : any ) = > {
if ( err . cancelled ) {
return err ;
}
createRangeQuery ( target : LokiQuery , options : RangeQueryOptions ) : LokiRangeQueryRequest {
const { query } = parseQuery ( target . expr ) ;
let range : { start? : number ; end? : number ; step? : number } = { } ;
if ( options . range && options . intervalMs ) {
const startNs = this . getTime ( options . range . from , false ) ;
const endNs = this . getTime ( options . range . to , true ) ;
const rangeMs = Math . ceil ( ( endNs - startNs ) / 1 e6 ) ;
const step = this . adjustInterval ( options . intervalMs , rangeMs ) / 1000 ;
const alignedTimes = {
start : startNs - ( startNs % 1 e9 ) ,
end : endNs + ( 1 e9 - ( endNs % 1 e9 ) ) ,
} ;
const error : DataQueryError = this . processError ( err , query ) ;
throw error ;
} )
) . pipe (
range = {
start : alignedTimes.start ,
end : alignedTimes.end ,
step ,
} ;
}
return {
. . . DEFAULT_QUERY_PARAMS ,
. . . range ,
query ,
limit : Math.min ( options . maxDataPoints || Infinity , this . maxLines ) ,
} ;
}
/ * *
* Attempts to send a query to / loki / api / v1 / query_range but falls back to the legacy endpoint if necessary .
* /
runRangeQueryWithFallback = (
target : LokiQuery ,
options : RangeQueryOptions ,
responseListLength = 1
) : Observable < DataQueryResponse > = > {
if ( target . liveStreaming ) {
return this . runLiveQuery ( target , options ) ;
}
const query = this . createRangeQuery ( target , options ) ;
return this . _request ( RANGE_QUERY_ENDPOINT , query ) . pipe (
catchError ( ( err : any ) = > this . throwUnless ( err , err . cancelled || err . status === 404 , target ) ) ,
filter ( ( response : any ) = > ( response . cancelled ? false : true ) ) ,
map ( ( response : any ) = > {
const data = this . processResult ( response . data , query ) ;
return { data , key : query.refId } ;
} )
switchMap ( ( response : { data : LokiResponse ; status : number } ) = >
iif < DataQueryResponse , DataQueryResponse > (
( ) = > response . status === 404 ,
defer ( ( ) = > this . runLegacyQuery ( target , options ) ) ,
defer ( ( ) = >
processRangeQueryResponse ( response . data , target , query , responseListLength , this . maxLines , options . reverse )
)
)
)
) ;
} ;
query ( options : DataQueryRequest < LokiQuery > ) : Observable < DataQueryResponse > {
const subQueries = options . targets
. filter ( target = > target . expr && ! target . hide )
. map ( target = > {
if ( target . liveStreaming ) {
return this . runLiveQuery ( options , target ) ;
}
return this . runQuery ( options , target ) ;
} ) ;
createLegacyLiveTarget ( target : LokiQuery , options : { maxDataPoints? : number } ) : LegacyTarget {
const { query , regexp } = parseQuery ( target . expr ) ;
const baseUrl = this . instanceSettings . url ;
const params = serializeParams ( { query } ) ;
// No valid targets, return the empty result to save a round trip.
if ( isEmpty ( subQueries ) ) {
return of ( {
data : [ ] ,
state : LoadingState.Done ,
} ) ;
}
return {
query ,
regexp ,
url : convertToWebSocketUrl ( ` ${ baseUrl } /api/prom/tail? ${ params } ` ) ,
refId : target.refId ,
size : Math.min ( options . maxDataPoints || Infinity , this . maxLines ) ,
} ;
}
return merge ( . . . subQueries ) ;
createLiveTarget ( target : LokiQuery , options : { maxDataPoints? : number } ) : LegacyTarget {
const { query , regexp } = parseQuery ( target . expr ) ;
const baseUrl = this . instanceSettings . url ;
const params = serializeParams ( { query } ) ;
return {
query ,
regexp ,
url : convertToWebSocketUrl ( ` ${ baseUrl } /loki/api/v1/tail? ${ params } ` ) ,
refId : target.refId ,
size : Math.min ( options . maxDataPoints || Infinity , this . maxLines ) ,
} ;
}
/ * *
* Runs live queries which in this case means creating a websocket and listening on it for new logs .
* This returns a bit different dataFrame than runQueries as it returns single dataframe even if there are multiple
* Loki streams , sets only common labels on dataframe . labels and has additional dataframe . fields . labels for unique
* labels per row .
* /
runLiveQuery = ( target : LokiQuery , options : { maxDataPoints? : number } ) : Observable < DataQueryResponse > = > {
const liveTarget = this . createLiveTarget ( target , options ) ;
return from ( this . getVersion ( ) ) . pipe (
mergeMap ( version = >
iif (
( ) = > version === 'v1' ,
defer ( ( ) = > this . streams . getStream ( liveTarget ) ) ,
defer ( ( ) = > {
const legacyTarget = this . createLegacyLiveTarget ( target , options ) ;
return this . streams . getLegacyStream ( legacyTarget ) ;
} )
)
) ,
map ( data = > ( {
data ,
key : ` loki- ${ liveTarget . refId } ` ,
state : LoadingState.Streaming ,
} ) )
) ;
} ;
interpolateVariablesInQueries ( queries : LokiQuery [ ] ) : LokiQuery [ ] {
let expandedQueries = queries ;
if ( queries && queries . length > 0 ) {
expandedQueries = queries . map ( query = > {
const expandedQuery = {
. . . query ,
datasource : this.name ,
expr : this.templateSrv.replace ( query . expr , { } , this . interpolateQueryExpr ) ,
} ;
return expandedQuery ;
} ) ;
if ( queries && queries . length ) {
expandedQueries = queries . map ( query = > ( {
. . . query ,
datasource : this.name ,
expr : this.templateSrv.replace ( query . expr , { } , this . interpolateQueryExpr ) ,
} ) ) ;
}
return expandedQueries ;
}
@ -250,13 +389,11 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
return this . languageProvider . importQueries ( queries , originMeta . id ) ;
}
metadataRequest ( url : string , params? : any ) {
// HACK to get label values for {job=|}, will be replaced when implementing LokiQueryField
const apiUrl = url . replace ( 'v1' , 'prom' ) ;
return this . _request ( apiUrl , params , { silent : true } ) . then ( ( res : DataQueryResponse ) = > {
const data : any = { data : { data : res.data.values || [ ] } } ;
return data ;
} ) ;
async metadataRequest ( url : string , params? : Record < string , string > ) {
const res = await this . _request ( url , params , { silent : true } ) . toPromise ( ) ;
return {
data : { data : res.data.values || [ ] } ,
} ;
}
interpolateQueryExpr ( value : any , variable : any ) {
@ -288,6 +425,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
default :
break ;
}
const expression = formatQuery ( selector , parsed . regexp ) ;
return { . . . query , expr : expression } ;
}
@ -297,107 +435,140 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
}
getTime ( date : string | DateTime , roundUp : boolean ) {
if ( isString ( date ) ) {
if ( typeof date === 'string' ) {
date = dateMath . parse ( date , roundUp ) ;
}
return Math . ceil ( date . valueOf ( ) * 1 e6 ) ;
}
getLogRowContext = ( row : LogRowModel , options? : LokiContextQueryOptions ) = > {
const target = this . prepareLogRowContextQueryTarget (
row ,
( options && options . limit ) || 10 ,
( options && options . direction ) || 'BACKWARD'
) ;
const reverse = options && options . direction === 'FORWARD' ;
return this . _request ( RANGE_QUERY_ENDPOINT , target )
. pipe (
catchError ( ( err : any ) = > {
if ( err . status === 404 ) {
return of ( err ) ;
}
const error : DataQueryError = {
message : 'Error during context query. Please check JS console logs.' ,
status : err.status ,
statusText : err.statusText ,
} ;
throw error ;
} ) ,
switchMap ( ( res : { data : LokiStreamResponse ; status : number } ) = >
iif (
( ) = > res . status === 404 ,
this . _request ( LEGACY_QUERY_ENDPOINT , target ) . pipe (
catchError ( ( err : any ) = > {
const error : DataQueryError = {
message : 'Error during context query. Please check JS console logs.' ,
status : err.status ,
statusText : err.statusText ,
} ;
throw error ;
} ) ,
map ( ( res : { data : LokiLegacyStreamResponse } ) = > ( {
data : res.data ? res . data . streams . map ( stream = > legacyLogStreamToDataFrame ( stream , reverse ) ) : [ ] ,
} ) )
) ,
of ( {
data : res.data ? res . data . data . result . map ( stream = > lokiStreamResultToDataFrame ( stream , reverse ) ) : [ ] ,
} )
)
)
)
. toPromise ( ) ;
} ;
prepareLogRowContextQueryTarget = ( row : LogRowModel , limit : number , direction : 'BACKWARD' | 'FORWARD' ) = > {
const query = Object . keys ( row . labels )
. map ( label = > {
return ` ${ label } =" ${ row . labels [ label ] } " ` ;
} )
. map ( label = > ` ${ label } =" ${ row . labels [ label ] } " ` )
. join ( ',' ) ;
const contextTimeBuffer = 2 * 60 * 60 * 1000 * 1 e6 ; // 2h buffer
const timeEpochNs = row . timeEpochMs * 1 e6 ;
const commontTargetOptons = {
const commonTargetOptions = {
limit ,
query : ` { ${ query } } ` ,
expr : ` { ${ query } } ` ,
direction ,
} ;
if ( direction === 'BACKWARD' ) {
return {
. . . commontTargetOptons ,
. . . commonTargetOpti ons ,
start : timeEpochNs - contextTimeBuffer ,
end : row.timestamp , // using RFC3339Nano format to avoid precision loss
end : timeEpochNs , // using RFC3339Nano format to avoid precision loss
direction ,
} ;
} else {
return {
. . . commontTargetOptons ,
start : row.timestamp , // start param in Loki API is inclusive so we'll have to filter out the row that this request is based from
. . . commonTargetOpti ons ,
start : timeEpochNs , // start param in Loki API is inclusive so we'll have to filter out the row that this request is based from
end : timeEpochNs + contextTimeBuffer ,
} ;
}
} ;
getLogRowContext = async ( row : LogRowModel , options? : LokiContextQueryOptions ) = > {
const target = this . prepareLogRowContextQueryTarget (
row ,
( options && options . limit ) || 10 ,
( options && options . direction ) || 'BACKWARD'
) ;
const series : DataFrame [ ] = [ ] ;
try {
const reverse = options && options . direction === 'FORWARD' ;
const result = await this . _request ( '/api/prom/query' , target ) ;
if ( result . data ) {
for ( const stream of result . data . streams || [ ] ) {
series . push ( logStreamToDataFrame ( stream , reverse ) ) ;
}
}
return {
data : series ,
} ;
} catch ( e ) {
const error : DataQueryError = {
message : 'Error during context query. Please check JS console logs.' ,
status : e.status ,
statusText : e.statusText ,
} ;
throw error ;
}
} ;
testDatasource() {
// Consider only last 10 minutes otherwise request takes too long
const startMs = Date . now ( ) - 10 * 60 * 1000 ;
const start = ` ${ startMs } 000000 ` ; // API expects nanoseconds
return this . _request ( '/api/prom/label' , { start } )
. then ( ( res : DataQueryResponse ) = > {
if ( res && res . data && res . data . values && res . data . values . length > 0 ) {
return { status : 'success' , message : 'Data source connected and labels found.' } ;
}
return {
status : 'error' ,
message :
'Data source connected, but no labels received. Verify that Loki and Promtail is configured properly.' ,
} ;
} )
. catch ( ( err : any ) = > {
let message = 'Loki: ' ;
if ( err . statusText ) {
message += err . statusText ;
} else {
message += 'Cannot connect to Loki' ;
}
return this . _request ( '/loki/api/v1/label' , { start } )
. pipe (
catchError ( ( err : any ) = > {
if ( err . status === 404 ) {
return of ( err ) ;
}
if ( err . status ) {
message += ` . ${ err . status } ` ;
}
throw err ;
} ) ,
switchMap ( ( response : { data : { values : string [ ] } ; status : number } ) = >
iif < DataQueryResponse , DataQueryResponse > (
( ) = > response . status === 404 ,
defer ( ( ) = > this . _request ( '/api/prom/label' , { start } ) ) ,
defer ( ( ) = > of ( response ) )
)
) ,
map ( res = >
res && res . data && res . data . values && res . data . values . length
? { status : 'success' , message : 'Data source connected and labels found.' }
: {
status : 'error' ,
message :
'Data source connected, but no labels received. Verify that Loki and Promtail is configured properly.' ,
}
) ,
catchError ( ( err : any ) = > {
let message = 'Loki: ' ;
if ( err . statusText ) {
message += err . statusText ;
} else {
message += 'Cannot connect to Loki' ;
}
if ( err . data && err . data . message ) {
message += ` . ${ err . data . message } ` ;
} else if ( err . data ) {
message += ` . ${ err . data } ` ;
}
return { status : 'error' , message : message } ;
} ) ;
if ( err . status ) {
message += ` . ${ err . status } ` ;
}
if ( err . data && err . data . message ) {
message += ` . ${ err . data . message } ` ;
} else if ( err . data ) {
message += ` . ${ err . data } ` ;
}
return of ( { status : 'error' , message : message } ) ;
} )
)
. toPromise ( ) ;
}
async annotationQuery ( options : AnnotationQueryRequest < LokiQuery > ) : Promise < AnnotationEvent [ ] > {
@ -405,8 +576,8 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
return [ ] ;
}
const request = queryRequestFromAnnotationOptions ( options ) ;
const { data } = await this . runQuery ( request , request . targets [ 0 ] ) . toPromise ( ) ;
const query = { refId : ` annotation- ${ options . annotation . name } ` , expr : options.annotation.expr } ;
const { data } = await this . runRangeQueryWithFallback ( query , options ) . toPromise ( ) ;
const annotations : AnnotationEvent [ ] = [ ] ;
for ( const frame of data ) {
@ -474,29 +645,48 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
dataFrame . fields = [ . . . dataFrame . fields , . . . Object . values ( fields ) ] ;
}
}
}
function queryRequestFromAnnotationOptions ( options : AnnotationQueryRequest < LokiQuery > ) : DataQueryRequest < LokiQuery > {
const refId = ` annotation- ${ options . annotation . name } ` ;
const target : LokiQuery = { refId , expr : options.annotation.expr } ;
return {
requestId : refId ,
range : options.range ,
targets : [ target ] ,
dashboardId : options.dashboard.id ,
scopedVars : null ,
startTime : Date.now ( ) ,
// This should mean the default defined on datasource is used.
maxDataPoints : 0 ,
// Dummy values, are required in type but not used here.
timezone : 'utc' ,
panelId : 0 ,
interval : '' ,
intervalMs : 0 ,
throwUnless = ( err : any , condition : boolean , target : LokiQuery ) = > {
if ( condition ) {
return of ( err ) ;
}
const error : DataQueryError = this . processError ( err , target ) ;
throw error ;
} ;
processError = ( err : any , target : LokiQuery ) : DataQueryError = > {
const error : DataQueryError = {
message : ( err && err . statusText ) || 'Unknown error during query transaction. Please check JS console logs.' ,
refId : target.refId ,
} ;
if ( err . data ) {
if ( typeof err . data === 'string' ) {
error . message = err . data ;
} else if ( err . data . error ) {
error . message = safeStringifyValue ( err . data . error ) ;
}
} else if ( err . message ) {
error . message = err . message ;
} else if ( typeof err === 'string' ) {
error . message = err ;
}
error . status = err . status ;
error . statusText = err . statusText ;
return error ;
} ;
adjustInterval ( interval : number , range : number ) {
// Loki will drop queries that might return more than 11000 data points.
// Calibrate interval if it is too small.
if ( interval !== 0 && range / interval > 11000 ) {
interval = Math . ceil ( range / 11000 ) ;
}
return Math . max ( interval , 1000 ) ;
}
}
export function lokiRegularEscape ( value : any ) {
@ -514,3 +704,7 @@ export function lokiSpecialRegexEscape(value: any) {
}
export default LokiDatasource ;
function isTimeSeries ( data : any ) : data is TimeSeries {
return data . hasOwnProperty ( 'datapoints' ) ;
}
kay delaney
6 years ago
committed by