package handlers import ( "context" stderrors "errors" "io" "net" "net/http" "net/url" "regexp" "strconv" "strings" "github.com/gin-gonic/gin" "github.com/prometheus/prometheus/model/labels" promparser "github.com/prometheus/prometheus/promql/parser" clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api" "github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto" apierrors "github.com/pingcap-inc/clinic/clinic_data_service/pkg/errors" "github.com/pingcap-inc/clinic/clinic_data_service/pkg/observability" "github.com/pingcap-inc/clinic/clinic_data_service/pkg/service" ) const canonicalTelemetryRequestMaxBytes = 64 << 10 var publicForbiddenMetrics = func() map[string]struct{} { names := []string{ "autoscale_rcu_batch_scale_total", "autoscale_rcu_pool_total", "autoscale_rcu_run_total", "autoscale_rcu_scale_down_total", "autoscale_rcu_scale_up_total", "autoscale_tikv_action_total", "autoscale_tikv_active_scalers", "autoscale_tikv_cooldown_skip_total", "autoscale_tikv_decision_total", "aws_ebs_csi_ec2_exceeded_iops_seconds_total", "aws_ebs_csi_ec2_exceeded_tp_seconds_total", "aws_ebs_csi_exceeded_tp_seconds_total", "aws_ebs_csi_read_bytes_total", "aws_ebs_csi_read_io_latency_seconds_bucket", "aws_ebs_csi_write_bytes_total", "aws_ebs_csi_write_io_latency_seconds_bucket", "container_cpu_cfs_periods_total", "container_cpu_cfs_throttled_periods_total", "container_cpu_usage_seconds_total", "container_memory_working_set_bytes", "instance:node_cpu_utilisation:rate1m", "instance:node_load1_per_cpu:ratio", "instance:node_memory_utilisation:ratio", "kube_daemonset_status_current_number_scheduled", "kube_daemonset_status_updated_number_scheduled", "kube_deployment_spec_replicas", "kube_horizontalpodautoscaler_spec_max_replicas", "kube_horizontalpodautoscaler_spec_min_replicas", "kube_horizontalpodautoscaler_spec_target_metric", "kube_horizontalpodautoscaler_status_current_replicas", "kube_horizontalpodautoscaler_status_desired_replicas", "kube_horizontalpodautoscaler_status_target_metric", "kube_node_info", "kube_node_labels", "kube_node_status_allocatable", "kube_persistentvolumeclaim_info", "kube_persistentvolumeclaim_resource_requests_storage_bytes", "kube_pod_container_info", "kube_pod_container_resource_limit_cpu_cores", "kube_pod_container_resource_limits", "kube_pod_container_resource_limits_memory_bytes", "kube_pod_container_resource_requests", "kube_pod_container_status_last_terminated_reason", "kube_pod_container_status_restarts_total", "kube_pod_status_phase", "kube_pod_status_ready", "kube_statefulset_replicas", "kubelet_volume_stats_inodes", "kubelet_volume_stats_inodes_used", "kubelet_volume_stats_used_bytes", "node_disk_reads_merged_total", "node_disk_writes_merged_total", "node_ethtool_bw_in_allowance_exceeded", "node_ethtool_bw_out_allowance_exceeded", "node_ethtool_conntrack_allowance_exceeded", "node_ethtool_info", "node_ethtool_linklocal_allowance_exceeded", "node_ethtool_pps_allowance_exceeded", "node_filesystem_avail_bytes", "node_filesystem_size_bytes", "node_icmp_rtt_ms", "node_icmp_rtt_seconds_bucket", "node_load1", "node_load15", "node_load5", "node_namespace_pod_container:container_cpu_usage_seconds_total:sum_rate", "node_netstat_Tcp_ActiveOpens", "node_netstat_Tcp_CurrEstab", "node_netstat_Tcp_PassiveOpens", "node_netstat_Tcp_RetransSegs", "node_network_receive_bytes_total", "node_network_receive_drop_total", "node_network_receive_packets_total", "node_network_transmit_bytes_total", "node_network_transmit_drop_total", "node_network_transmit_packets_total", "node_nf_conntrack_entries", "node_nf_conntrack_entries_limit", "node_nf_conntrack_stat_drop", "node_nf_conntrack_stat_early_drop", "node_nf_conntrack_stat_insert_failed", "node_sockstat_TCP_inuse", "node_uname_info", "kube_node_annotations", "kube_node_spec_taint", "node_dmi_info", "node_cpu_info", "karpenter_pods_state", "karpenter_cloudprovider_instance_type_price_estimate", "karpenter_cloudprovider_instance_type_offering_price_estimate", "karpenter_cloudprovider_instance_type_offering_available", "karpenter_cloudprovider_instance_type_memory_bytes", "karpenter_cloudprovider_instance_type_cpu_cores", "karpenter_nodepools_cost_total", "node_cpu_hourly_cost", "node_gpu_hourly_cost", "node_ram_hourly_cost", "node_total_hourly_cost", "kubecost_node_is_spot", "kube_storageclass_annotations", "kube_storageclass_info", "kube_storageclass_labels", "kube_persistentvolume_annotations", "kube_persistentvolume_info", "kube_persistentvolume_labels", "kube_persistentvolumeclaim_annotations", "kube_persistentvolumeclaim_labels", "kube_volumeattachment_labels", "kube_volumeattachment_status_attachment_metadata", "node_disk_info", "node_nvme_info", "pv_hourly_cost", } forbidden := make(map[string]struct{}, len(names)) for _, name := range names { forbidden[name] = struct{}{} } return forbidden }() func forbiddenMetricsInPromQL(query string) bool { expr, err := promparser.ParseExpr(query) if err != nil { return false } forbidden := false promparser.Inspect(expr, func(node promparser.Node, _ []promparser.Node) error { selector, ok := node.(*promparser.VectorSelector) if !ok { return nil } if _, ok := publicForbiddenMetrics[selector.Name]; ok { forbidden = true return nil } for _, matcher := range selector.LabelMatchers { if matcher.Name == "__name__" && matcher.Type == labels.MatchEqual { if _, ok := publicForbiddenMetrics[matcher.Value]; ok { forbidden = true } } } return nil }) return forbidden } func forbiddenMetricsInSeries(matchParams []string) bool { for _, match := range matchParams { if forbiddenMetricsInPromQL(match) { return true } } return false } func emptyPrometheusQueryResponse(resultType string) *dto.PrometheusResponse { return &dto.PrometheusResponse{ Status: "success", Data: map[string]any{ "resultType": resultType, "result": []any{}, }, } } func emptyPrometheusSeriesResponse() *dto.PrometheusResponse { return &dto.PrometheusResponse{Status: "success", Data: []any{}} } type canonicalTelemetryResolver struct { ownership canonicalClusterOwnershipReader target canonicalClusterTargetReader } func (r canonicalTelemetryResolver) resolve( c *gin.Context, rec *observability.DataProxyRecord, orgID uint64, clusterID string, ) (*clinic_internal_api.ClusterDataURLs, bool) { if r.ownership == nil || r.target == nil { rec.ErrCode = "TELEMETRY_UNAVAILABLE" writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "telemetry service is temporarily unavailable", nil) return nil, false } cluster, err := r.ownership.GetByOrgAndClusterID(c.Request.Context(), orgID, clusterID) if err != nil { writeCanonicalOwnershipError(c, rec, err) return nil, false } expectedOrgID := strconv.FormatUint(orgID, 10) if cluster == nil || cluster.ClusterID != clusterID || cluster.OrgID != expectedOrgID { rec.ErrCode = "OWNERSHIP_IDENTITY_MISMATCH" writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster ownership service returned an invalid response", nil) return nil, false } target, err := r.target.GetCanonicalClusterDataURLs(c.Request.Context(), clusterID) if err != nil { writeCanonicalTargetError(c, rec, err) return nil, false } if target == nil || target.ClusterID != clusterID || target.OrgID != expectedOrgID { rec.ErrCode = "TARGET_IDENTITY_MISMATCH" writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster target service returned an invalid response", nil) return nil, false } rec.Provider = strings.ToLower(strings.TrimSpace(target.Provider)) rec.Region = strings.TrimSpace(target.Region) rec.DeployTypeV2 = target.DeployTypeV2 return target, true } // CanonicalMetricsHandler adds org-scoped REST aliases over the existing // Prometheus-compatible metrics execution core. type CanonicalMetricsHandler struct { legacy *MetricsQueryHandler resolver canonicalTelemetryResolver } func NewCanonicalMetricsHandler( legacy *MetricsQueryHandler, ownership canonicalClusterOwnershipReader, target canonicalClusterTargetReader, ) *CanonicalMetricsHandler { return &CanonicalMetricsHandler{legacy: legacy, resolver: canonicalTelemetryResolver{ownership: ownership, target: target}} } // Query executes an instant PromQL query. // // @Summary Execute a cluster instant metrics query // @ID canonical-metrics-query-get // @Tags data-proxy-metrics // @Accept x-www-form-urlencoded // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param query query string true "PromQL expression" // @Success 200 {object} dto.PrometheusResponse // @Failure 400 {object} dto.PrometheusResponse // @Failure 422 {object} dto.PrometheusResponse // @Failure 502 {object} dto.PrometheusResponse // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/metrics/query [get] func (h *CanonicalMetricsHandler) Query(c *gin.Context) { h.handle(c, "canonical_metrics_instant", func(clusterID string, target *clinic_internal_api.ClusterDataURLs) { h.legacy.instantQuery(c, "canonical_metrics_instant", clusterID, target) }) } // QueryForm executes the same instant-query core for a form POST. // // @Summary Execute a cluster instant metrics query // @ID canonical-metrics-query-post // @Tags data-proxy-metrics // @Accept x-www-form-urlencoded // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param query formData string true "PromQL expression" // @Success 200 {object} dto.PrometheusResponse // @Failure 400 {object} dto.PrometheusResponse // @Failure 422 {object} dto.PrometheusResponse // @Failure 502 {object} dto.PrometheusResponse // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/metrics/query [post] func (h *CanonicalMetricsHandler) QueryForm(c *gin.Context) { h.Query(c) } // QueryRange executes a range PromQL query. // // @Summary Execute a cluster range metrics query // @ID canonical-metrics-query-range-get // @Tags data-proxy-metrics // @Accept x-www-form-urlencoded // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param query query string true "PromQL expression" // @Success 200 {object} dto.PrometheusResponse // @Failure 400 {object} dto.PrometheusResponse // @Failure 422 {object} dto.PrometheusResponse // @Failure 502 {object} dto.PrometheusResponse // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/metrics/query-range [get] func (h *CanonicalMetricsHandler) QueryRange(c *gin.Context) { h.handle(c, "canonical_metrics_range", func(clusterID string, target *clinic_internal_api.ClusterDataURLs) { h.legacy.rangeQuery(c, "canonical_metrics_range", clusterID, target) }) } func (h *CanonicalMetricsHandler) PublicQueryRange(c *gin.Context) { h.handle(c, "canonical_metrics_range", func(clusterID string, target *clinic_internal_api.ClusterDataURLs) { h.legacy.publicRangeQuery(c, clusterID, target) }) } // QueryRangeForm executes the same range-query core for a form POST. // // @Summary Execute a cluster range metrics query // @ID canonical-metrics-query-range-post // @Tags data-proxy-metrics // @Accept x-www-form-urlencoded // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param query formData string true "PromQL expression" // @Param start formData string true "Range start" // @Param end formData string true "Range end" // @Param step formData string true "Query resolution step" // @Success 200 {object} dto.PrometheusResponse // @Failure 400 {object} dto.PrometheusResponse // @Failure 422 {object} dto.PrometheusResponse // @Failure 502 {object} dto.PrometheusResponse // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/metrics/query-range [post] func (h *CanonicalMetricsHandler) QueryRangeForm(c *gin.Context) { h.QueryRange(c) } // Series returns matching Prometheus series. // // @Summary List matching cluster metrics series // @ID canonical-metrics-series-get // @Tags data-proxy-metrics // @Accept x-www-form-urlencoded // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param match[] query []string true "Series selectors" // @Success 200 {object} dto.PrometheusResponse // @Failure 400 {object} dto.PrometheusResponse // @Failure 422 {object} dto.PrometheusResponse // @Failure 502 {object} dto.PrometheusResponse // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/metrics/series [get] func (h *CanonicalMetricsHandler) Series(c *gin.Context) { h.handle(c, "canonical_metrics_series", func(clusterID string, target *clinic_internal_api.ClusterDataURLs) { h.legacy.series(c, "canonical_metrics_series", clusterID, target) }) } func (h *CanonicalMetricsHandler) PublicSeries(c *gin.Context) { h.handle(c, "canonical_metrics_series", func(clusterID string, target *clinic_internal_api.ClusterDataURLs) { h.legacy.publicSeries(c, clusterID, target) }) } // SeriesForm executes the same series-query core for a form POST. // // @Summary List matching cluster metrics series // @ID canonical-metrics-series-post // @Tags data-proxy-metrics // @Accept x-www-form-urlencoded // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param match[] formData []string true "Series selectors" // @Success 200 {object} dto.PrometheusResponse // @Failure 400 {object} dto.PrometheusResponse // @Failure 422 {object} dto.PrometheusResponse // @Failure 502 {object} dto.PrometheusResponse // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/metrics/series [post] func (h *CanonicalMetricsHandler) SeriesForm(c *gin.Context) { h.Series(c) } func (h *CanonicalMetricsHandler) handle(c *gin.Context, endpoint string, execute func(string, *clinic_internal_api.ClusterDataURLs)) { start, rec := beginCanonicalDataProxyRequest(c, endpoint) rec.QueryType = endpoint failed := true defer func() { if failed { observability.RecordDataProxy(c, start, rec) } }() if !validateCanonicalTelemetryTransport(c, rec) || !rejectCanonicalPlacement(c, rec) { return } orgID, clusterID, ok := parseCanonicalTelemetryScope(c, rec) if !ok { return } if h == nil || h.legacy == nil || h.legacy.service == nil { rec.ErrCode = "METRICS_UNAVAILABLE" writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "metrics service is temporarily unavailable", nil) return } target, ok := h.resolver.resolve(c, rec, orgID, clusterID) if !ok { return } failed = false execute(clusterID, target) } // RegisterRoutes registers only the additive canonical metrics aliases. func (h *CanonicalMetricsHandler) RegisterRoutes(r *gin.Engine) { group := r.Group("/data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/metrics") group.GET("/query", h.Query) group.POST("/query", h.QueryForm) group.GET("/query-range", h.QueryRange) group.POST("/query-range", h.QueryRangeForm) group.GET("/series", h.Series) group.POST("/series", h.SeriesForm) // Public TiDB Cloud aliases use the dbaas-auth Data Proxy namespace. publicGroup := r.Group("/api/v1/orgs/:orgId/data-proxy/metrics/:clusterId") publicGroup.GET("/query_range", h.PublicQueryRange) publicGroup.GET("/series", h.PublicSeries) } type canonicalLokiProxy interface { ProxyRequest(context.Context, string, *url.URL, string, io.Reader, http.Header, *service.LokiClusterInfo) (*http.Response, error) } type canonicalBYOCLokiProxy interface { ProxyBYOCRequest(context.Context, string, *url.URL, string, io.Reader, http.Header, *service.LokiClusterInfo) (*http.Response, error) } // CanonicalBusinessLogsHandler adds org-scoped REST aliases over the existing // Loki rewrite/proxy core. type CanonicalBusinessLogsHandler struct { proxy canonicalLokiProxy resolver canonicalTelemetryResolver } func NewCanonicalBusinessLogsHandler( proxy canonicalLokiProxy, ownership canonicalClusterOwnershipReader, target canonicalClusterTargetReader, ) *CanonicalBusinessLogsHandler { return &CanonicalBusinessLogsHandler{proxy: proxy, resolver: canonicalTelemetryResolver{ownership: ownership, target: target}} } // QueryRange proxies a cluster LogQL range query. // // @Summary Execute a cluster log range query // @ID canonical-logs-query-range // @Tags data-proxy-logs // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param query query string true "LogQL expression" // @Success 200 {object} map[string]interface{} // @Failure 400 {object} apierrors.APIError // @Failure 502 {object} apierrors.APIError // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/logs/query-range [get] func (h *CanonicalBusinessLogsHandler) QueryRange(c *gin.Context) { h.proxyRequest(c, "/loki/api/v1/query_range", "canonical_logs_query_range") } // Labels proxies Loki label discovery. // // @Summary List cluster log labels // @ID canonical-logs-labels // @Tags data-proxy-logs // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Success 200 {object} map[string]interface{} // @Failure 400 {object} apierrors.APIError // @Failure 502 {object} apierrors.APIError // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/logs/labels [get] func (h *CanonicalBusinessLogsHandler) Labels(c *gin.Context) { h.proxyRequest(c, "/loki/api/v1/labels", "canonical_logs_labels") } // LabelValues proxies Loki label-value discovery. // // @Summary List cluster log label values // @ID canonical-logs-label-values // @Tags data-proxy-logs // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param name path string true "Loki label name" // @Success 200 {object} map[string]interface{} // @Failure 400 {object} apierrors.APIError // @Failure 502 {object} apierrors.APIError // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/logs/labels/{name}/values [get] func (h *CanonicalBusinessLogsHandler) LabelValues(c *gin.Context) { name := c.Param("name") if !canonicalLokiLabelName.MatchString(name) { start, rec := beginCanonicalDataProxyRequest(c, "canonical_logs_label_values") rec.QueryType = "loki" rec.ErrCode = "INVALID_LABEL_NAME" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "invalid Loki label name", nil) observability.RecordDataProxy(c, start, rec) return } h.proxyRequest(c, "/loki/api/v1/label/"+name+"/values", "canonical_logs_label_values") } // Series proxies Loki stream discovery. // // @Summary List matching cluster log streams // @ID canonical-logs-series // @Tags data-proxy-logs // @Produce json // @Param orgId path string true "TiDB Cloud/DBAAS org ID" // @Param clusterId path string true "Cluster ID" // @Param match[] query []string true "LogQL stream selectors" // @Success 200 {object} map[string]interface{} // @Failure 400 {object} apierrors.APIError // @Failure 502 {object} apierrors.APIError // @Security BearerAuth // @Router /data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/logs/series [get] func (h *CanonicalBusinessLogsHandler) Series(c *gin.Context) { h.proxyRequest(c, "/loki/api/v1/series", "canonical_logs_series") } func (h *CanonicalBusinessLogsHandler) proxyRequest(c *gin.Context, lokiPath, endpoint string) { start, rec := beginCanonicalDataProxyRequest(c, endpoint) rec.QueryType = "loki" defer func() { observability.RecordDataProxy(c, start, rec) }() if !validateCanonicalTelemetryTransport(c, rec) || !rejectCanonicalPlacement(c, rec) { return } orgID, clusterID, ok := parseCanonicalTelemetryScope(c, rec) if !ok { return } if h == nil || h.proxy == nil { rec.ErrCode = "LOGS_UNAVAILABLE" writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "logs service is temporarily unavailable", nil) return } target, ok := h.resolver.resolve(c, rec, orgID, clusterID) if !ok { return } if strings.TrimSpace(target.LokiURL) == "" { rec.ErrCode = "LOKI_URL_NOT_CONFIGURED" writeResourceError(c, http.StatusNotFound, "NotFound", "logs are not available for this cluster", nil) return } info := &service.LokiClusterInfo{ ClusterID: target.ClusterID, ParentID: target.ParentID, DeployTypeV2: target.DeployTypeV2, ProjectID: target.ProjectID, ClusterMetricsURL: target.ClusterMetricsURL, LokiURL: target.LokiURL, } byoc := service.IsBYOCLogsCluster(target.DeployType, target.DeployTypeV2) var resp *http.Response var err error if byoc { info.TenantID = target.ClinicTenantID info.Provider = target.Provider info.Region = target.Region info.DeployType = target.DeployType byocProxy, ok := h.proxy.(canonicalBYOCLokiProxy) if !ok { rec.ErrCode = "LOGS_UNAVAILABLE" writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "logs service is temporarily unavailable", nil) return } resp, err = byocProxy.ProxyBYOCRequest(c.Request.Context(), lokiPath, c.Request.URL, c.Request.Method, c.Request.Body, canonicalLokiHeaders(c.Request.Header), info) } else { resp, err = h.proxy.ProxyRequest(c.Request.Context(), lokiPath, c.Request.URL, c.Request.Method, c.Request.Body, canonicalLokiHeaders(c.Request.Header), info) } if err != nil { rec.ErrCode = "LOKI_PROXY_FAILED" if byoc && stderrors.Is(err, service.ErrInvalidBYOCLogQuery) { writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "invalid BYOC logs query", nil) return } writeCanonicalLogsProxyError(c, err) return } defer resp.Body.Close() if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { rec.ErrCode = "LOKI_UPSTREAM_ERROR" _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 32<<10)) writeCanonicalLogsUpstreamError(c, resp.StatusCode) return } for key, values := range resp.Header { for _, value := range values { c.Writer.Header().Add(key, value) } } c.Header("X-Loki-Proxy", "clinic-data-service") c.Status(resp.StatusCode) if _, err := io.Copy(c.Writer, resp.Body); err != nil { rec.ErrCode = "LOKI_RESPONSE_COPY_FAILED" } } // RegisterRoutes registers only the additive canonical business-log aliases. func (h *CanonicalBusinessLogsHandler) RegisterRoutes(r *gin.Engine) { group := r.Group("/data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/logs") group.GET("/query-range", h.QueryRange) group.GET("/labels", h.Labels) group.GET("/labels/:name/values", h.LabelValues) group.GET("/series", h.Series) // Public TiDB Cloud alias. Keep the established query_range spelling for // compatibility with the Prometheus/Loki API surface exposed by Data Proxy. publicGroup := r.Group("/api/v1/orgs/:orgId/data-proxy/loki/:clusterId") publicGroup.GET("/query_range", h.QueryRange) } var canonicalLokiLabelName = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]{0,254}$`) func parseCanonicalTelemetryScope(c *gin.Context, rec *observability.DataProxyRecord) (uint64, string, bool) { orgID, ok := parseOrgID(c, rec) if !ok { return 0, "", false } clusterID := c.Param("clusterId") if err := validateResourceID(clusterID); err != nil { rec.ErrCode = "INVALID_CLUSTER_ID" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", err.Error(), nil) return 0, "", false } return orgID, clusterID, true } var canonicalPlacementKeys = map[string]struct{}{ "org": {}, "orgid": {}, "organizationid": {}, "oid": {}, "cluster": {}, "clusterid": {}, "cid": {}, "provider": {}, "region": {}, "datasource": {}, "datasourceuid": {}, "uid": {}, "metricsurl": {}, "prometheusurl": {}, "clustermetricsurl": {}, "dataplanemetricsurl": {}, "lokiurl": {}, "controlplaneinfo": {}, "projectid": {}, "parentid": {}, "deploytype": {}, "deploytypev2": {}, "keyspaceid": {}, } var canonicalPlacementHeaders = map[string]struct{}{ "xorgid": {}, "xclusterid": {}, "xprovider": {}, "xregion": {}, "xdatasource": {}, "xmetricsurl": {}, "xprometheusurl": {}, "xlokiurl": {}, "xcontrolplaneinfo": {}, "xprojectid": {}, "xparentid": {}, } func rejectCanonicalPlacement(c *gin.Context, rec *observability.DataProxyRecord) bool { query, err := url.ParseQuery(c.Request.URL.RawQuery) if err != nil { rec.ErrCode = "INVALID_QUERY_PARAMETERS" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "query parameters are malformed", nil) return false } for key := range query { if _, blocked := canonicalPlacementKeys[normalizeCanonicalRoutingKey(key)]; blocked { return writeCanonicalPlacementError(c, rec) } } if c.Request.Method == http.MethodPost && strings.HasPrefix(strings.ToLower(c.GetHeader("Content-Type")), "application/x-www-form-urlencoded") { if err := c.Request.ParseForm(); err != nil { var maxBytesError *http.MaxBytesError if stderrors.As(err, &maxBytesError) { rec.ErrCode = "PAYLOAD_TOO_LARGE" writeResourceError(c, http.StatusRequestEntityTooLarge, "PayloadTooLarge", "request body exceeds the allowed size", nil) return false } rec.ErrCode = "INVALID_FORM" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "invalid form body", nil) return false } for key := range c.Request.PostForm { if _, blocked := canonicalPlacementKeys[normalizeCanonicalRoutingKey(key)]; blocked { return writeCanonicalPlacementError(c, rec) } } } for key := range c.Request.Header { if _, blocked := canonicalPlacementHeaders[normalizeCanonicalRoutingKey(key)]; blocked { return writeCanonicalPlacementError(c, rec) } } return true } func writeCanonicalPlacementError(c *gin.Context, rec *observability.DataProxyRecord) bool { rec.ErrCode = "CALLER_ROUTING_REJECTED" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "cluster routing and placement are server controlled", nil) return false } func normalizeCanonicalRoutingKey(key string) string { key = strings.TrimSuffix(strings.ToLower(key), "[]") return strings.NewReplacer("-", "", "_", "", ".", "").Replace(key) } func validateCanonicalTelemetryTransport(c *gin.Context, rec *observability.DataProxyRecord) bool { if len(c.Request.URL.RawQuery) > canonicalTelemetryRequestMaxBytes { rec.ErrCode = "QUERY_PARAMETERS_TOO_LARGE" writeResourceError(c, http.StatusRequestEntityTooLarge, "PayloadTooLarge", "query parameters exceed the allowed size", nil) return false } if _, err := url.ParseQuery(c.Request.URL.RawQuery); err != nil { rec.ErrCode = "INVALID_QUERY_PARAMETERS" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "query parameters are malformed", nil) return false } if c.Request.Method == http.MethodGet { if c.Request.ContentLength != 0 || len(c.Request.TransferEncoding) != 0 { rec.ErrCode = "UNSUPPORTED_REQUEST_BODY" writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "request body is not supported", nil) return false } return true } if c.Request.Method != http.MethodPost { rec.ErrCode = "UNSUPPORTED_METHOD" writeResourceError(c, http.StatusMethodNotAllowed, "MethodNotAllowed", "request method is not supported", nil) return false } mediaType := strings.ToLower(strings.TrimSpace(strings.SplitN(c.GetHeader("Content-Type"), ";", 2)[0])) if mediaType != "application/x-www-form-urlencoded" { rec.ErrCode = "UNSUPPORTED_CONTENT_TYPE" writeResourceError(c, http.StatusUnsupportedMediaType, "UnsupportedMediaType", "content type must be application/x-www-form-urlencoded", nil) return false } c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, canonicalTelemetryRequestMaxBytes) return true } var canonicalLokiHeaderAllowlist = map[string]struct{}{ "Accept": {}, "Accept-Encoding": {}, "Content-Type": {}, "User-Agent": {}, "X-Request-Id": {}, "X-Trace-Id": {}, "Traceparent": {}, "Tracestate": {}, } func canonicalLokiHeaders(source http.Header) http.Header { result := make(http.Header) for key := range canonicalLokiHeaderAllowlist { for _, value := range source.Values(key) { result.Add(key, value) } } return result } func writeCanonicalLogsProxyError(c *gin.Context, err error) { var networkError net.Error if stderrors.Is(err, context.DeadlineExceeded) || (stderrors.As(err, &networkError) && networkError.Timeout()) { writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "logs upstream request timed out", nil) return } writeResourceError(c, http.StatusBadGateway, "BadGateway", "logs upstream request failed", nil) } func writeCanonicalLogsUpstreamError(c *gin.Context, status int) { switch status { case http.StatusTooManyRequests: writeResourceError(c, http.StatusTooManyRequests, "ResourceExhausted", "logs upstream rate limit exceeded", nil) case http.StatusRequestTimeout, http.StatusGatewayTimeout: writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "logs upstream request timed out", nil) default: writeResourceError(c, http.StatusBadGateway, "BadGateway", "logs upstream request failed", nil) } } var ( _ canonicalLokiProxy = (*service.LokiProxyService)(nil) _ dto.PrometheusResponse _ apierrors.APIError )