@ -32,7 +32,6 @@ import (
"github.com/grafana/loki/v3/pkg/util/constants"
"github.com/grafana/loki/v3/pkg/util/extract"
util_log "github.com/grafana/loki/v3/pkg/util/log"
"github.com/grafana/loki/v3/pkg/util/spanlogger"
)
var (
@ -319,15 +318,13 @@ func (c *IndexReaderWriter) chunksToSeries(ctx context.Context, in []logproto.Ch
func ( c * IndexReaderWriter ) LabelNamesForMetricName ( ctx context . Context , userID string , from , through model . Time , metricName string , matchers ... * labels . Matcher ) ( [ ] string , error ) {
sp , ctx := opentracing . StartSpanFromContext ( ctx , "SeriesStore.LabelNamesForMetricName" )
defer sp . Finish ( )
log := spanlogger . FromContext ( ctx )
defer log . Span . Finish ( )
// Fetch the series IDs from the index
seriesIDs , err := c . lookupSeriesByMetricNameMatchers ( ctx , from , through , userID , metricName , matchers )
if err != nil {
return nil , err
}
level . Debug ( log ) . Log ( "series-ids" , len ( seriesIDs ) )
sp . LogKV ( "series-ids" , len ( seriesIDs ) )
// Lookup the series in the index to get label names.
labelNames , err := c . lookupLabelNamesBySeries ( ctx , from , through , userID , seriesIDs )
@ -336,10 +333,10 @@ func (c *IndexReaderWriter) LabelNamesForMetricName(ctx context.Context, userID
if err == series_index . ErrNotSupported {
return c . lookupLabelNamesByChunks ( ctx , from , through , userID , seriesIDs )
}
level . Error ( log ) . Log ( "msg" , "lookupLabelNamesBySeries" , "err" , err )
sp . LogKV ( "msg" , "lookupLabelNamesBySeries" , "err" , err )
return nil , err
}
level . Debug ( log ) . Log ( "labelNames" , len ( labelNames ) )
sp . LogKV ( "labelNames" , len ( labelNames ) )
return labelNames , nil
}
@ -347,14 +344,12 @@ func (c *IndexReaderWriter) LabelNamesForMetricName(ctx context.Context, userID
func ( c * IndexReaderWriter ) LabelValuesForMetricName ( ctx context . Context , userID string , from , through model . Time , metricName string , labelName string , matchers ... * labels . Matcher ) ( [ ] string , error ) {
sp , ctx := opentracing . StartSpanFromContext ( ctx , "SeriesStore.LabelValuesForMetricName" )
defer sp . Finish ( )
log := spanlogger . FromContext ( ctx )
defer log . Span . Finish ( )
if len ( matchers ) != 0 {
return c . labelValuesForMetricNameWithMatchers ( ctx , userID , from , through , metricName , labelName , matchers ... )
}
level . Debug ( log ) . Log ( "from" , from , "through" , through , "metricName" , metricName , "labelName" , labelName )
sp . LogKV ( "from" , from , "through" , through , "metricName" , metricName , "labelName" , labelName )
queries , err := c . schema . GetReadQueriesForMetricLabel ( from , through , userID , metricName , labelName )
if err != nil {
@ -634,10 +629,8 @@ func (c *IndexReaderWriter) lookupEntriesByQueries(ctx context.Context, queries
func ( c * IndexReaderWriter ) lookupLabelNamesBySeries ( ctx context . Context , from , through model . Time , userID string , seriesIDs [ ] string ) ( [ ] string , error ) {
sp , ctx := opentracing . StartSpanFromContext ( ctx , "SeriesStore.lookupLabelNamesBySeries" )
defer sp . Finish ( )
log := spanlogger . FromContext ( ctx )
defer log . Span . Finish ( )
level . Debug ( log ) . Log ( "seriesIDs" , len ( seriesIDs ) )
sp . LogKV ( "seriesIDs" , len ( seriesIDs ) )
queries := make ( [ ] series_index . Query , 0 , len ( seriesIDs ) )
for _ , seriesID := range seriesIDs {
qs , err := c . schema . GetLabelNamesForSeries ( from , through , userID , [ ] byte ( seriesID ) )
@ -646,7 +639,7 @@ func (c *IndexReaderWriter) lookupLabelNamesBySeries(ctx context.Context, from,
}
queries = append ( queries , qs ... )
}
level . Debug ( log ) . Log ( "queries" , len ( queries ) )
sp . LogKV ( "queries" , len ( queries ) )
entries := entriesPool . Get ( ) . ( * [ ] series_index . Entry )
defer entriesPool . Put ( entries )
err := c . lookupEntriesByQueries ( ctx , queries , entries )
@ -654,7 +647,7 @@ func (c *IndexReaderWriter) lookupLabelNamesBySeries(ctx context.Context, from,
return nil , err
}
level . Debug ( log ) . Log ( "entries" , len ( * entries ) )
sp . LogKV ( "entries" , len ( * entries ) )
var result util . UniqueStrings
for _ , entry := range * entries {
@ -671,34 +664,32 @@ func (c *IndexReaderWriter) lookupLabelNamesBySeries(ctx context.Context, from,
func ( c * IndexReaderWriter ) lookupLabelNamesByChunks ( ctx context . Context , from , through model . Time , userID string , seriesIDs [ ] string ) ( [ ] string , error ) {
sp , ctx := opentracing . StartSpanFromContext ( ctx , "SeriesStore.lookupLabelNamesByChunks" )
defer sp . Finish ( )
log := spanlogger . FromContext ( ctx )
defer log . Span . Finish ( )
// Lookup the series in the index to get the chunks.
chunkIDs , err := c . lookupChunksBySeries ( ctx , from , through , userID , seriesIDs )
if err != nil {
level . Error ( log ) . Log ( "msg" , "lookupChunksBySeries" , "err" , err )
sp . LogKV ( "msg" , "lookupChunksBySeries" , "err" , err )
return nil , err
}
level . Debug ( log ) . Log ( "chunk-ids" , len ( chunkIDs ) )
sp . LogKV ( "chunk-ids" , len ( chunkIDs ) )
chunks , err := c . convertChunkIDsToChunks ( ctx , userID , chunkIDs )
if err != nil {
level . Error ( log ) . Log ( "err" , "convertChunkIDsToChunks" , "err" , err )
sp . LogKV ( "err" , "convertChunkIDsToChunks" , "err" , err )
return nil , err
}
// Filter out chunks that are not in the selected time range and keep a single chunk per fingerprint
filtered := filterChunksByTime ( from , through , chunks )
filtered = filterChunksByUniqueFingerprint ( filtered )
level . Debug ( log ) . Log ( "Chunks post filtering" , len ( chunks ) )
sp . LogKV ( "Chunks post filtering" , len ( chunks ) )
chunksPerQuery . Observe ( float64 ( len ( filtered ) ) )
// Now fetch the actual chunk data from Memcache / S3
allChunks , err := c . fetcher . FetchChunks ( ctx , filtered )
if err != nil {
level . Error ( log ) . Log ( "msg" , "FetchChunks" , "err" , err )
sp . LogKV ( "msg" , "FetchChunks" , "err" , err )
return nil , err
}
return labelNamesFromChunks ( allChunks ) , nil
Dylan Guedes
2 years ago
committed by