package prom import ( "fmt" "regexp" "strconv" "time" ) const DefaultBeylaJobSelector = `job=~".*beyla.*|.*alloy.*"` type RequestSource string const ( RequestSourceBeyla RequestSource = "beyla" RequestSourceIstio RequestSource = "istio" ) type RequestQueries struct { Rate string Errors string P50 string P95 string HistogramCoverage string StatusCoverage string ObservedPods string Population string HistogramUniformity string } var beylaJobMatcher = regexp.MustCompile(`^job\s*(=|=~)\s*("(?:[^"\\]|\\.)*")$`) func BuildRequestQueries(step time.Duration, sel PodSelection, scope WorkloadMetricsScope, source RequestSource, jobSelector string) (RequestQueries, error) { cluster, err := scope.Matchers() if err != nil { return RequestQueries{}, err } return buildRequestQueries(step, sel, cluster, source, jobSelector, nil, nil) } func BuildIdentityRequestQueries(step time.Duration, sel PodSelection, pods []WorkloadPodIdentity, source RequestSource, jobSelector string) (RequestQueries, error) { branches, err := buildWorkloadIdentityFilter(pods, source) if err != nil { return RequestQueries{}, err } return buildRequestQueries(step, sel, "", source, jobSelector, branches, nil) } func BuildHistoryRequestQueries(step time.Duration, history WorkloadHistory, source RequestSource, jobSelector string) (RequestQueries, error) { if _, err := history.OwnerQuery(); err != nil { return RequestQueries{}, err } cluster, _ := history.Scope.Matchers() if jobSelector == "" { jobSelector = `job=~".+"` } return buildRequestQueries(step, PodSelection{Namespace: history.Namespace}, cluster, source, jobSelector, nil, &history) } func buildRequestQueries(step time.Duration, sel PodSelection, cluster string, source RequestSource, jobSelector string, branches *workloadIdentityFilter, history *WorkloadHistory) (RequestQueries, error) { if history == nil && sel.IsEmpty() { return RequestQueries{}, fmt.Errorf("no current pods") } var metric, target, status, podLabel string unitScale := "" switch source { case RequestSourceBeyla: if jobSelector == "" { jobSelector = DefaultBeylaJobSelector } match := beylaJobMatcher.FindStringSubmatch(jobSelector) if match == nil { return RequestQueries{}, fmt.Errorf("the configured Beyla job filter must be one job equality or regex matcher for workload charts; Live Traffic accepts wider fragments") } value, err := strconv.Unquote(match[2]) if err != nil { return RequestQueries{}, fmt.Errorf("invalid Beyla job matcher: %w", err) } if match[1] == "=~" { if _, err := regexp.Compile(value); err != nil { return RequestQueries{}, fmt.Errorf("invalid Beyla job regex: %w", err) } } metric = "http_server_request_duration_seconds" target = fmt.Sprintf(`%s,k8s_namespace_name=%s,k8s_pod_name=~'%s'`, jobSelector, strconv.Quote(sel.Namespace), exactSetPattern(sel.Pods)) if history != nil { target = fmt.Sprintf(`%s,k8s_namespace_name=%s,k8s_pod_name!=""`, jobSelector, strconv.Quote(sel.Namespace)) } status = "http_response_status_code" podLabel = "k8s_pod_name" case RequestSourceIstio: metric = "istio_request_duration_milliseconds" // A destination reporter's scrape Pod is the observed Pod. A waypoint's // scrape Pod is the proxy, not its target; it cannot use this attribution. target = fmt.Sprintf(`reporter="destination",request_protocol="http",destination_workload_namespace=%s,namespace=%s%s`, strconv.Quote(sel.Namespace), strconv.Quote(sel.Namespace), scopeClause(sel)) if history != nil { target = fmt.Sprintf(`reporter="destination",request_protocol="http",destination_workload_namespace=%s,namespace=%s,destination_workload=%s`, strconv.Quote(sel.Namespace), strconv.Quote(sel.Namespace), strconv.Quote(history.Name)) } status = "response_code" podLabel = "pod" unitScale = " / 1000" default: return RequestQueries{}, fmt.Errorf("unsupported request source %q", source) } expression := func(metric, extra string) string { if history != nil { query, _ := history.expression(metric, target+extra, WorkloadRateWindow(step).String(), source == RequestSourceBeyla) return query } return workloadMetricExpression(metric, cluster, target+extra, branches, WorkloadRateWindow(step).String()) } countMetric := metric + "_count" if source == RequestSourceIstio { countMetric = "istio_requests_total" } rate := func(metric, extra string) string { return "sum(max without (job,instance) (" + expression(metric, extra) + "))" } total := rate(countMetric, "") errors := rate(countMetric, fmt.Sprintf(`,%s=~"5.."`, status)) bucketRates := "max without (job,instance) (" + expression(metric+"_bucket", "") + ")" buckets := "sum by (le) (" + bucketRates + ")" bucketPopulation := "count by (le) (" + bucketRates + ")" quantile := func(q string) string { return "histogram_quantile(" + q + ", " + buckets + ")" + unitScale } histogramCoverage := rate(metric+"_bucket", `,le="+Inf"`) populationMetrics := expression(countMetric, "") + " or " + expression(metric+"_bucket", "") if source == RequestSourceIstio { populationMetrics += " or " + expression(metric+"_count", "") counter := "max without (job,instance) (" + expression(countMetric, "") + ")" histogramCount := "max without (job,instance) (" + expression(metric+"_count", "") + ")" infinity := "max without (job,instance,le) (" + expression(metric+"_bucket", `,le="+Inf"`) + ")" // Envoy merges histograms asynchronously from request counters. Match // their complete label populations, but compare values within the histogram. population := "count(" + counter + " and " + histogramCount + ") == bool count(" + counter + " or " + histogramCount + ")" consistent := "count(" + histogramCount + " == " + infinity + ") == bool count(" + histogramCount + " or " + infinity + ")" noForeignBuckets := "((count(" + bucketRates + " unless ignoring(le) " + histogramCount + ") or vector(0)) == bool 0)" histogramCoverage = "(" + population + ") * (" + consistent + ") * " + noForeignBuckets } return RequestQueries{ Rate: total, Errors: errors, P50: quantile("0.50"), P95: quantile("0.95"), HistogramCoverage: histogramCoverage, HistogramUniformity: "min(" + bucketPopulation + ") / max(" + bucketPopulation + ")", StatusCoverage: rate(countMetric, fmt.Sprintf(`,%s=~"0|[1-5][0-9][0-9]"`, status)), ObservedPods: "count(count by (" + podLabel + ") (" + expression(countMetric, "") + "))", Population: "count(count by (job,replica,prometheus_replica,__replica__) (" + populationMetrics + "))", }, nil }