| 107 | } |
| 108 | |
| 109 | func NewQueryTripperware( |
| 110 | log log.Logger, |
| 111 | registerer prometheus.Registerer, |
| 112 | forwardHeaders []string, |
| 113 | queryRangeMiddleware []Middleware, |
| 114 | instantRangeMiddleware []Middleware, |
| 115 | queryRangeCodec Codec, |
| 116 | instantQueryCodec Codec, |
| 117 | limits Limits, |
| 118 | queryAnalyzer querysharding.Analyzer, |
| 119 | defaultSubQueryInterval time.Duration, |
| 120 | maxSubQuerySteps int64, |
| 121 | lookbackDelta time.Duration, |
| 122 | ) Tripperware { |
| 123 | |
| 124 | // Per tenant query metrics. |
| 125 | queriesPerTenant := promauto.With(registerer).NewCounterVec(prometheus.CounterOpts{ |
| 126 | Name: "cortex_query_frontend_queries_total", |
| 127 | Help: "Total queries sent per tenant.", |
| 128 | }, []string{"op", "source", "user"}) |
| 129 | |
| 130 | rejectedQueriesPerTenant := promauto.With(registerer).NewCounterVec(prometheus.CounterOpts{ |
| 131 | Name: "cortex_query_frontend_rejected_queries_total", |
| 132 | Help: "Total rejected queries per tenant.", |
| 133 | }, []string{"op", "user"}) |
| 134 | |
| 135 | activeUsers := users.NewActiveUsersCleanupWithDefaultValues(func(user string) { |
| 136 | err := util.DeleteMatchingLabels(queriesPerTenant, map[string]string{"user": user}) |
| 137 | if err != nil { |
| 138 | level.Warn(log).Log("msg", "failed to remove cortex_query_frontend_queries_total metric for user", "user", user) |
| 139 | } |
| 140 | }) |
| 141 | |
| 142 | // Start cleanup. If cleaner stops or fail, we will simply not clean the metrics for inactive users. |
| 143 | _ = activeUsers.StartAsync(context.Background()) |
| 144 | return func(next http.RoundTripper) http.RoundTripper { |
| 145 | // Finally, if the user selected any query middleware, stitch it in. |
| 146 | if len(queryRangeMiddleware) > 0 || len(instantRangeMiddleware) > 0 { |
| 147 | queryrange := NewRoundTripper(next, queryRangeCodec, forwardHeaders, queryRangeMiddleware...) |
| 148 | instantQuery := NewRoundTripper(next, instantQueryCodec, forwardHeaders, instantRangeMiddleware...) |
| 149 | return RoundTripFunc(func(r *http.Request) (*http.Response, error) { |
| 150 | isQuery := strings.HasSuffix(r.URL.Path, "/query") |
| 151 | isQueryRange := strings.HasSuffix(r.URL.Path, "/query_range") |
| 152 | isSeries := strings.HasSuffix(r.URL.Path, "/series") |
| 153 | isRemoteRead := strings.HasSuffix(r.URL.Path, "/read") |
| 154 | isLabelNames := strings.HasSuffix(r.URL.Path, "/labels") |
| 155 | isLabelValues := strings.HasSuffix(r.URL.Path, "/values") |
| 156 | isMetadata := strings.HasSuffix(r.URL.Path, "/metadata") |
| 157 | isQueryExemplars := strings.HasSuffix(r.URL.Path, "/query_exemplars") |
| 158 | isFormatQuery := strings.HasSuffix(r.URL.Path, "/format_query") |
| 159 | isParseQuery := strings.HasSuffix(r.URL.Path, "/parse_query") |
| 160 | |
| 161 | op := opTypeQuery |
| 162 | switch { |
| 163 | case isQueryRange: |
| 164 | op = opTypeQueryRange |
| 165 | case isSeries: |
| 166 | op = opTypeSeries |