hanzoai/o11y
2
0
Fork 0
mirror of https://github.com/hanzoai/o11y.git synced 2026-10-15 12:58:33 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity
Files
main
o11y/pkg/query-service/app/http_handler.go
T
zeekay 6c3c881330 Read exceptions from the event fact table 
listErrors and its siblings read a retired table.

Claude-Session: https://claude.ai/code/session_011i2AG9nwpiKUCVxAdDzYuU
2026-09-29 19:33:41 -07:00

3996 lines
121 KiB
Go
Raw Permalink Blame History

package app
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/flagger"
"github.com/hanzoai/o11y/pkg/modules/thirdpartyapi"
"github.com/hanzoai/o11y/pkg/queryparser"
"log/slog"
"math"
"net/http"
"regexp"
"slices"
"sort"
"strconv"
"strings"
"sync"
"text/template"
"time"
"github.com/prometheus/prometheus/promql"
"github.com/hanzoai/o11y/pkg/http/render"
"github.com/hanzoai/o11y/pkg/licensing"
"github.com/hanzoai/o11y/pkg/o11y"
"github.com/hanzoai/o11y/pkg/query-service/app/integrations"
"github.com/hanzoai/o11y/pkg/types/retentiontypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/gorilla/websocket"
jsoniter "github.com/json-iterator/go"
"github.com/hanzoai/o11y/pkg/contextlinks"
traceFunnelsModule "github.com/hanzoai/o11y/pkg/modules/tracefunnel"
"github.com/hanzoai/o11y/pkg/query-service/agentConf"
"github.com/hanzoai/o11y/pkg/query-service/app/inframetrics"
queues2 "github.com/hanzoai/o11y/pkg/query-service/app/integrations/messagingQueues/queues"
"github.com/hanzoai/o11y/pkg/query-service/app/logs"
logsv3 "github.com/hanzoai/o11y/pkg/query-service/app/logs/v3"
logsv4 "github.com/hanzoai/o11y/pkg/query-service/app/logs/v4"
"github.com/hanzoai/o11y/pkg/query-service/app/metrics"
metricsv3 "github.com/hanzoai/o11y/pkg/query-service/app/metrics/v3"
"github.com/hanzoai/o11y/pkg/query-service/app/querier"
querierV2 "github.com/hanzoai/o11y/pkg/query-service/app/querier/v2"
"github.com/hanzoai/o11y/pkg/query-service/app/queryBuilder"
tracesV3 "github.com/hanzoai/o11y/pkg/query-service/app/traces/v3"
tracesV4 "github.com/hanzoai/o11y/pkg/query-service/app/traces/v4"
"github.com/hanzoai/o11y/pkg/query-service/constants"
v3 "github.com/hanzoai/o11y/pkg/query-service/model/v3"
"github.com/hanzoai/o11y/pkg/query-service/postprocess"
"github.com/hanzoai/o11y/pkg/types/authtypes"
"github.com/hanzoai/o11y/pkg/types/coretypes"
"github.com/hanzoai/o11y/pkg/types/ctxtypes"
"github.com/hanzoai/o11y/pkg/types/featuretypes"
"github.com/hanzoai/o11y/pkg/types/instrumentationtypes"
"github.com/hanzoai/o11y/pkg/types/opamptypes"
"github.com/hanzoai/o11y/pkg/types/pipelinetypes"
qbtypes "github.com/hanzoai/o11y/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/hanzoai/o11y/pkg/types/ruletypes"
traceFunnels "github.com/hanzoai/o11y/pkg/types/tracefunneltypes"
"github.com/hanzoai/o11y/pkg/query-service/app/integrations/messagingQueues/kafka"
"github.com/hanzoai/o11y/pkg/query-service/app/logparsingpipeline"
"github.com/hanzoai/o11y/pkg/query-service/interfaces"
"github.com/hanzoai/o11y/pkg/query-service/model"
"github.com/hanzoai/o11y/pkg/ruler"
"github.com/hanzoai/o11y/pkg/version"
)
type status string
const (
statusSuccess status = "success"
statusError status = "error"
)
// APIHandler implements the query service public API
type APIHandler struct {
logger *slog.Logger
reader interfaces.Reader
ruleManager ruler.Ruler
querier interfaces.Querier
querierV2 interfaces.Querier
queryBuilder *queryBuilder.QueryBuilder
// temporalityMap is a map of metric name to temporality to avoid fetching
// temporality for the same metric multiple times.
//
// Querying the v4 table on a low cardinal temporality column should be
// fast, but we can still avoid the query if we have the data in memory.
temporalityMap map[string]map[v3.Temporality]bool
temporalityMux sync.Mutex
IntegrationsController *integrations.Controller
LogsParsingPipelineController *logparsingpipeline.LogParsingPipelineController
// SetupCompleted indicates if O11y is ready for general use.
// at the moment, we mark the app ready when the first user
// is registers.
SetupCompleted bool
// Websocket connection upgrader
Upgrader *websocket.Upgrader
hostsRepo *inframetrics.HostsRepo
processesRepo *inframetrics.ProcessesRepo
podsRepo *inframetrics.PodsRepo
nodesRepo *inframetrics.NodesRepo
namespacesRepo *inframetrics.NamespacesRepo
clustersRepo *inframetrics.ClustersRepo
// workloads
deploymentsRepo *inframetrics.DeploymentsRepo
daemonsetsRepo *inframetrics.DaemonSetsRepo
statefulsetsRepo *inframetrics.StatefulSetsRepo
jobsRepo *inframetrics.JobsRepo
pvcsRepo *inframetrics.PvcsRepo
LicensingAPI licensing.API
QueryParserAPI *queryparser.API
O11y *o11y.O11y
}
type APIHandlerOpts struct {
// business data reader e.g. datastore
Reader interfaces.Reader
// Integrations
IntegrationsController *integrations.Controller
// Log parsing pipelines
LogsParsingPipelineController *logparsingpipeline.LogParsingPipelineController
// Flux Interval
FluxInterval time.Duration
LicensingAPI licensing.API
QueryParserAPI *queryparser.API
O11y *o11y.O11y
}
// NewAPIHandler returns an APIHandler
func NewAPIHandler(opts APIHandlerOpts, config o11y.Config) (*APIHandler, error) {
querierOpts := querier.QuerierOptions{
Reader: opts.Reader,
Cache: opts.O11y.Cache,
KeyGenerator: queryBuilder.NewKeyGenerator(),
FluxInterval: opts.FluxInterval,
}
querierOptsV2 := querierV2.QuerierOptions{
Reader: opts.Reader,
Cache: opts.O11y.Cache,
KeyGenerator: queryBuilder.NewKeyGenerator(),
FluxInterval: opts.FluxInterval,
}
querier := querier.NewQuerier(querierOpts)
querierv2 := querierV2.NewQuerier(querierOptsV2)
hostsRepo := inframetrics.NewHostsRepo(opts.Reader, querierv2)
processesRepo := inframetrics.NewProcessesRepo(opts.Reader, querierv2)
podsRepo := inframetrics.NewPodsRepo(opts.Reader, querierv2)
nodesRepo := inframetrics.NewNodesRepo(opts.Reader, querierv2)
namespacesRepo := inframetrics.NewNamespacesRepo(opts.Reader, querierv2)
clustersRepo := inframetrics.NewClustersRepo(opts.Reader, querierv2)
deploymentsRepo := inframetrics.NewDeploymentsRepo(opts.Reader, querierv2)
daemonsetsRepo := inframetrics.NewDaemonSetsRepo(opts.Reader, querierv2)
statefulsetsRepo := inframetrics.NewStatefulSetsRepo(opts.Reader, querierv2)
jobsRepo := inframetrics.NewJobsRepo(opts.Reader, querierv2)
pvcsRepo := inframetrics.NewPvcsRepo(opts.Reader, querierv2)
//quickFilterModule := quickfilter.NewAPI(opts.QuickFilterModule)
aH := &APIHandler{
logger: slog.Default(),
reader: opts.Reader,
temporalityMap: make(map[string]map[v3.Temporality]bool),
ruleManager: opts.O11y.Ruler,
IntegrationsController: opts.IntegrationsController,
LogsParsingPipelineController: opts.LogsParsingPipelineController,
querier: querier,
querierV2: querierv2,
hostsRepo: hostsRepo,
processesRepo: processesRepo,
podsRepo: podsRepo,
nodesRepo: nodesRepo,
namespacesRepo: namespacesRepo,
clustersRepo: clustersRepo,
deploymentsRepo: deploymentsRepo,
daemonsetsRepo: daemonsetsRepo,
statefulsetsRepo: statefulsetsRepo,
jobsRepo: jobsRepo,
pvcsRepo: pvcsRepo,
LicensingAPI: opts.LicensingAPI,
O11y: opts.O11y,
QueryParserAPI: opts.QueryParserAPI,
}
logsQueryBuilder := logsv4.PrepareLogsQuery
tracesQueryBuilder := tracesV4.PrepareTracesQuery
builderOpts := queryBuilder.QueryBuilderOptions{
BuildMetricQuery: metricsv3.PrepareMetricQuery,
BuildTraceQuery: tracesQueryBuilder,
BuildLogQuery: logsQueryBuilder,
}
aH.queryBuilder = queryBuilder.NewQueryBuilder(builderOpts)
// Zero-onboarding: o11y holds no native identity or first-run setup. Identity
// is asserted by the Hanzo IAM session (pkg/identn/iamidentn) and the tenant
// org is auto-provisioned on first request, so o11y is always "set up" — there
// is no first-user/registration gate and the data APIs are on from boot.
aH.SetupCompleted = true
aH.Upgrader = &websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool {
return true
},
}
return aH, nil
}
// todo(remove): Implemented at render package (github.com/hanzoai/o11y/pkg/http/render) with the new error structure
type structuredResponse struct {
Data interface{} `json:"data"`
Total int `json:"total"`
Limit int `json:"limit"`
Offset int `json:"offset"`
Errors []structuredError `json:"errors"`
}
// todo(remove): Implemented at render package (github.com/hanzoai/o11y/pkg/http/render) with the new error structure
type structuredError struct {
Code int `json:"code,omitempty"`
Msg string `json:"msg"`
}
// todo(remove): Implemented at render package (github.com/hanzoai/o11y/pkg/http/render) with the new error structure
type ApiResponse struct {
Status status `json:"status"`
Data interface{} `json:"data,omitempty"`
ErrorType model.ErrorType `json:"errorType,omitempty"`
Error string `json:"error,omitempty"`
}
// toApiError translates a pkg/errors typed error into the legacy
// model.ApiError to preserve the v1 JSON response shape.
func toApiError(err error) *model.ApiError {
t, _, _, _, _, _ := errors.Unwrapb(err)
var typ model.ErrorType
switch t {
case errors.TypeInvalidInput:
typ = model.ErrorBadData
case errors.TypeNotFound:
typ = model.ErrorNotFound
case errors.TypeAlreadyExists:
typ = model.ErrorConflict
case errors.TypeUnauthenticated:
typ = model.ErrorUnauthorized
case errors.TypeForbidden:
typ = model.ErrorForbidden
case errors.TypeUnsupported:
typ = model.ErrorNotImplemented
case errors.TypeTimeout:
typ = model.ErrorTimeout
case errors.TypeCanceled:
typ = model.ErrorCanceled
default:
typ = model.ErrorInternal
}
return &model.ApiError{Typ: typ, Err: err}
}
// todo(remove): Implemented at render package (github.com/hanzoai/o11y/pkg/http/render) with the new error structure
func RespondError(w http.ResponseWriter, apiErr model.BaseApiError, data interface{}) {
json := jsoniter.ConfigCompatibleWithStandardLibrary
b, err := json.Marshal(&ApiResponse{
Status: statusError,
ErrorType: apiErr.Type(),
Error: apiErr.Error(),
Data: data,
})
if err != nil {
slog.Error("error marshalling json response", errors.Attr(err))
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
var code int
switch apiErr.Type() {
case model.ErrorBadData:
code = http.StatusBadRequest
case model.ErrorExec:
code = 422
case model.ErrorCanceled, model.ErrorTimeout:
code = http.StatusServiceUnavailable
case model.ErrorInternal:
code = http.StatusInternalServerError
case model.ErrorNotFound:
code = http.StatusNotFound
case model.ErrorNotImplemented:
code = http.StatusNotImplemented
case model.ErrorUnauthorized:
code = http.StatusUnauthorized
case model.ErrorForbidden:
code = http.StatusForbidden
case model.ErrorConflict:
code = http.StatusConflict
default:
code = http.StatusInternalServerError
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
if n, err := w.Write(b); err != nil {
slog.Error("error writing response", "bytes_written", n, errors.Attr(err))
}
}
// todo(remove): Implemented at render package (github.com/hanzoai/o11y/pkg/http/render) with the new error structure
func writeHttpResponse(w http.ResponseWriter, data interface{}) {
json := jsoniter.ConfigCompatibleWithStandardLibrary
b, err := json.Marshal(&ApiResponse{
Status: statusSuccess,
Data: data,
})
if err != nil {
slog.Error("error marshalling json response", errors.Attr(err))
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
if n, err := w.Write(b); err != nil {
slog.Error("error writing response", "bytes_written", n, errors.Attr(err))
}
}
// todo(remove): Implemented at render package (github.com/hanzoai/o11y/pkg/http/render) with the new error structure
func (aH *APIHandler) Respond(w http.ResponseWriter, data interface{}) {
writeHttpResponse(w, data)
}
func Intersection(a, b []int) (c []int) {
m := make(map[int]bool)
for _, item := range a {
m[item] = true
}
for _, item := range b {
if _, ok := m[item]; ok {
c = append(c, item)
}
}
return
}
// populateTemporality adds the temporality to the query if it is not present
func (aH *APIHandler) PopulateTemporality(ctx context.Context, orgID valuer.UUID, qp *v3.QueryRangeParamsV3) error {
aH.temporalityMux.Lock()
defer aH.temporalityMux.Unlock()
missingTemporality := make([]string, 0)
metricNameToTemporality := make(map[string]map[v3.Temporality]bool)
if qp.CompositeQuery != nil && len(qp.CompositeQuery.BuilderQueries) > 0 {
for _, query := range qp.CompositeQuery.BuilderQueries {
// if there is no temporality specified in the query but we have it in the map
// then use the value from the map
if query.Temporality == "" && aH.temporalityMap[query.AggregateAttribute.Key] != nil {
// We prefer delta if it is available
if aH.temporalityMap[query.AggregateAttribute.Key][v3.Delta] {
query.Temporality = v3.Delta
} else if aH.temporalityMap[query.AggregateAttribute.Key][v3.Cumulative] {
query.Temporality = v3.Cumulative
} else {
query.Temporality = v3.Unspecified
}
}
// we don't have temporality for this metric
if query.DataSource == v3.DataSourceMetrics && query.Temporality == "" {
missingTemporality = append(missingTemporality, query.AggregateAttribute.Key)
}
if _, ok := metricNameToTemporality[query.AggregateAttribute.Key]; !ok {
metricNameToTemporality[query.AggregateAttribute.Key] = make(map[v3.Temporality]bool)
}
}
}
nameToTemporality, err := aH.reader.FetchTemporality(ctx, orgID, missingTemporality)
if err != nil {
return err
}
if qp.CompositeQuery != nil && len(qp.CompositeQuery.BuilderQueries) > 0 {
for name := range qp.CompositeQuery.BuilderQueries {
query := qp.CompositeQuery.BuilderQueries[name]
if query.DataSource == v3.DataSourceMetrics && query.Temporality == "" {
if nameToTemporality[query.AggregateAttribute.Key][v3.Delta] {
query.Temporality = v3.Delta
} else if nameToTemporality[query.AggregateAttribute.Key][v3.Cumulative] {
query.Temporality = v3.Cumulative
} else {
query.Temporality = v3.Unspecified
}
aH.temporalityMap[query.AggregateAttribute.Key] = nameToTemporality[query.AggregateAttribute.Key]
}
}
}
return nil
}
func (aH *APIHandler) testRule(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
defer r.Body.Close()
body, err := io.ReadAll(r.Body)
if err != nil {
aH.logger.ErrorContext(r.Context(), "error reading request body for test rule", errors.Attr(err))
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
defer cancel()
alertCount, err := aH.ruleManager.TestNotification(ctx, orgID, string(body))
if err != nil {
RespondError(w, toApiError(err), nil)
return
}
response := map[string]interface{}{
"alertCount": alertCount,
"message": "notification sent",
}
aH.Respond(w, response)
}
func (aH *APIHandler) getRuleStats(w http.ResponseWriter, r *http.Request) {
ruleID := coretypes.Param(r, "id")
params := model.QueryRuleStateHistory{}
err := json.NewDecoder(r.Body).Decode(&params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
totalCurrentTriggers, err := aH.reader.GetTotalTriggers(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
currentTriggersSeries, err := aH.reader.GetTriggersByInterval(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
currentAvgResolutionTime, err := aH.reader.GetAvgResolutionTime(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
currentAvgResolutionTimeSeries, err := aH.reader.GetAvgResolutionTimeByInterval(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
if params.End-params.Start >= 86400000 {
days := int64(math.Ceil(float64(params.End-params.Start) / 86400000))
params.Start -= days * 86400000
params.End -= days * 86400000
} else {
params.Start -= 86400000
params.End -= 86400000
}
totalPastTriggers, err := aH.reader.GetTotalTriggers(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
pastTriggersSeries, err := aH.reader.GetTriggersByInterval(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
pastAvgResolutionTime, err := aH.reader.GetAvgResolutionTime(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
pastAvgResolutionTimeSeries, err := aH.reader.GetAvgResolutionTimeByInterval(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
if math.IsNaN(currentAvgResolutionTime) || math.IsInf(currentAvgResolutionTime, 0) {
currentAvgResolutionTime = 0
}
if math.IsNaN(pastAvgResolutionTime) || math.IsInf(pastAvgResolutionTime, 0) {
pastAvgResolutionTime = 0
}
stats := model.Stats{
TotalCurrentTriggers: totalCurrentTriggers,
TotalPastTriggers: totalPastTriggers,
CurrentTriggersSeries: currentTriggersSeries,
PastTriggersSeries: pastTriggersSeries,
CurrentAvgResolutionTime: strconv.FormatFloat(currentAvgResolutionTime, 'f', -1, 64),
PastAvgResolutionTime: strconv.FormatFloat(pastAvgResolutionTime, 'f', -1, 64),
CurrentAvgResolutionTimeSeries: currentAvgResolutionTimeSeries,
PastAvgResolutionTimeSeries: pastAvgResolutionTimeSeries,
}
aH.Respond(w, stats)
}
func (aH *APIHandler) getOverallStateTransitions(w http.ResponseWriter, r *http.Request) {
ruleID := coretypes.Param(r, "id")
params := model.QueryRuleStateHistory{}
err := json.NewDecoder(r.Body).Decode(&params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
stateItems, err := aH.reader.GetOverallStateTransitions(r.Context(), ruleID, &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, stateItems)
}
func (aH *APIHandler) getRuleStateHistory(w http.ResponseWriter, r *http.Request) {
idStr := coretypes.Param(r, "id")
id, err := valuer.NewUUID(idStr)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
params := model.QueryRuleStateHistory{}
err = json.NewDecoder(r.Body).Decode(&params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
if err := params.Validate(); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
res, err := aH.reader.ReadRuleStateHistoryByRuleID(r.Context(), id.StringValue(), &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
rule, err := aH.ruleManager.GetRule(r.Context(), id)
if err == nil {
for idx := range res.Items {
lbls := make(map[string]string)
err := json.Unmarshal([]byte(res.Items[idx].Labels), &lbls)
if err != nil {
continue
}
end := time.Unix(res.Items[idx].UnixMilli/1000, 0)
// why are we subtracting 3 minutes?
// the query range is calculated based on the rule's evalWindow and evalDelay
// alerts have 2 minutes delay built in, so we need to subtract that from the start time
// to get the correct query range
start := end.Add(-rule.EvalWindow.Duration() - 3*time.Minute)
if rule.AlertType == ruletypes.AlertTypeLogs {
// TODO(srikanthccv): re-visit this and support multiple queries
var q qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
for _, query := range rule.RuleCondition.CompositeQuery.Queries {
if query.Type == qbtypes.QueryTypeBuilder {
switch spec := query.Spec.(type) {
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
q = spec
}
}
}
filterExpr := ""
if q.Filter != nil && q.Filter.Expression != "" {
filterExpr = q.Filter.Expression
}
whereClause := contextlinks.PrepareFilterExpression(lbls, filterExpr, q.GroupBy)
res.Items[idx].RelatedLogsLink = contextlinks.PrepareParamsForLogsV5(start, end, whereClause).Encode()
} else if rule.AlertType == ruletypes.AlertTypeTraces {
// TODO(srikanthccv): re-visit this and support multiple queries
var q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]
for _, query := range rule.RuleCondition.CompositeQuery.Queries {
if query.Type == qbtypes.QueryTypeBuilder {
switch spec := query.Spec.(type) {
case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
q = spec
}
}
}
filterExpr := ""
if q.Filter != nil && q.Filter.Expression != "" {
filterExpr = q.Filter.Expression
}
whereClause := contextlinks.PrepareFilterExpression(lbls, filterExpr, q.GroupBy)
res.Items[idx].RelatedTracesLink = contextlinks.PrepareParamsForTracesV5(start, end, whereClause).Encode()
}
}
}
aH.Respond(w, res)
}
func (aH *APIHandler) getRuleStateHistoryTopContributors(w http.ResponseWriter, r *http.Request) {
idStr := coretypes.Param(r, "id")
id, err := valuer.NewUUID(idStr)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
params := model.QueryRuleStateHistory{}
err = json.NewDecoder(r.Body).Decode(&params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
res, err := aH.reader.ReadRuleStateHistoryTopContributorsByRuleID(r.Context(), id.StringValue(), &params)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, res)
}
func prepareQuery(r *http.Request) (string, error) {
var postData *model.DashboardVars
if err := json.NewDecoder(r.Body).Decode(&postData); err != nil {
return "", fmt.Errorf("failed to decode request body: %v", err)
}
query := strings.TrimSpace(postData.Query)
if query == "" {
return "", fmt.Errorf("query is required")
}
notAllowedOps := []string{
"alter table",
"drop table",
"truncate table",
"drop database",
"drop view",
"drop function",
}
for _, op := range notAllowedOps {
if strings.Contains(strings.ToLower(query), op) {
return "", fmt.Errorf("operation %s is not allowed", op)
}
}
vars := make(map[string]string)
for k, v := range postData.Variables {
vars[k] = metrics.FormattedValue(v)
}
tmpl := template.New("dashboard-vars")
tmpl, tmplErr := tmpl.Parse(query)
if tmplErr != nil {
return "", tmplErr
}
var queryBuf bytes.Buffer
tmplErr = tmpl.Execute(&queryBuf, vars)
if tmplErr != nil {
return "", tmplErr
}
if !constants.IsDotMetricsEnabled {
return queryBuf.String(), nil
}
query = queryBuf.String()
// Now handle $var replacements (simple string replace)
keys := make([]string, 0, len(vars))
for k := range vars {
keys = append(keys, k)
}
sort.Slice(keys, func(i, j int) bool {
return len(keys[i]) > len(keys[j])
})
newQuery := query
for _, k := range keys {
placeholder := "$" + k
v := vars[k]
newQuery = strings.ReplaceAll(newQuery, placeholder, v)
}
return newQuery, nil
}
func (aH *APIHandler) queryDashboardVarsV2(w http.ResponseWriter, r *http.Request) {
query, err := prepareQuery(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
dashboardVars, err := aH.reader.QueryDashboardVars(r.Context(), query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
aH.Respond(w, dashboardVars)
}
func (aH *APIHandler) queryRangeMetrics(w http.ResponseWriter, r *http.Request) {
query, apiErrorObj := parseQueryRangeRequest(r)
if apiErrorObj != nil {
RespondError(w, apiErrorObj, nil)
return
}
// TODO: add structured logging for query and apiError if needed
ctx := r.Context()
if to := r.FormValue("timeout"); to != "" {
var cancel context.CancelFunc
timeout, err := parseMetricsDuration(to)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
ctx, cancel = context.WithTimeout(ctx, timeout)
defer cancel()
}
res, qs, apiError := aH.reader.GetQueryRangeResult(ctx, query)
if apiError != nil {
RespondError(w, apiError, nil)
return
}
if res.Err != nil {
aH.logger.ErrorContext(r.Context(), "error in query range metrics", errors.Attr(res.Err))
}
if res.Err != nil {
switch res.Err.(type) {
case promql.ErrQueryCanceled:
RespondError(w, &model.ApiError{Typ: model.ErrorCanceled, Err: res.Err}, nil)
case promql.ErrQueryTimeout:
RespondError(w, &model.ApiError{Typ: model.ErrorTimeout, Err: res.Err}, nil)
}
RespondError(w, &model.ApiError{Typ: model.ErrorExec, Err: res.Err}, nil)
return
}
response_data := &model.QueryData{
ResultType: res.Value.Type(),
Result: res.Value,
Stats: qs,
}
aH.Respond(w, response_data)
}
func (aH *APIHandler) queryMetrics(w http.ResponseWriter, r *http.Request) {
queryParams, apiErrorObj := parseInstantQueryMetricsRequest(r)
if apiErrorObj != nil {
RespondError(w, apiErrorObj, nil)
return
}
// TODO: add structured logging for query and apiError if needed
ctx := r.Context()
if to := r.FormValue("timeout"); to != "" {
var cancel context.CancelFunc
timeout, err := parseMetricsDuration(to)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
ctx, cancel = context.WithTimeout(ctx, timeout)
defer cancel()
}
res, qs, apiError := aH.reader.GetInstantQueryMetricsResult(ctx, queryParams)
if apiError != nil {
RespondError(w, apiError, nil)
return
}
if res.Err != nil {
aH.logger.ErrorContext(r.Context(), "error in query range metrics", errors.Attr(res.Err))
}
if res.Err != nil {
switch res.Err.(type) {
case promql.ErrQueryCanceled:
RespondError(w, &model.ApiError{Typ: model.ErrorCanceled, Err: res.Err}, nil)
case promql.ErrQueryTimeout:
RespondError(w, &model.ApiError{Typ: model.ErrorTimeout, Err: res.Err}, nil)
}
RespondError(w, &model.ApiError{Typ: model.ErrorExec, Err: res.Err}, nil)
}
responseData := &model.QueryData{
ResultType: res.Value.Type(),
Result: res.Value,
Stats: qs,
}
aH.Respond(w, responseData)
}
func (aH *APIHandler) registerEvent(w http.ResponseWriter, r *http.Request) {
request, err := parseRegisterEventRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
claims, errv2 := authtypes.ClaimsFromContext(r.Context())
if errv2 == nil {
switch request.EventType {
case model.TrackEvent:
aH.O11y.Analytics.TrackUser(r.Context(), claims.OrgID, claims.IdentityID(), request.EventName, request.Attributes)
}
aH.WriteJSON(w, r, map[string]string{"data": "Event Processed Successfully"})
} else {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
}
}
func (aH *APIHandler) getUsage(w http.ResponseWriter, r *http.Request) {
query, err := parseGetUsageRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, err := aH.reader.GetUsage(r.Context(), query)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getServicesTopLevelOps(w http.ResponseWriter, r *http.Request) {
var start, end time.Time
var services []string
type topLevelOpsParams struct {
Service string `json:"service"`
Start string `json:"start"`
End string `json:"end"`
}
var params topLevelOpsParams
err := json.NewDecoder(r.Body).Decode(&params)
if err != nil {
aH.logger.ErrorContext(r.Context(), "error reading request body for get top operations", errors.Attr(err))
}
if params.Service != "" {
services = []string{params.Service}
}
startEpoch := params.Start
if startEpoch != "" {
startEpochInt, err := strconv.ParseInt(startEpoch, 10, 64)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, "Error reading start time")
return
}
start = time.Unix(0, startEpochInt)
}
endEpoch := params.End
if endEpoch != "" {
endEpochInt, err := strconv.ParseInt(endEpoch, 10, 64)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, "Error reading end time")
return
}
end = time.Unix(0, endEpochInt)
}
result, apiErr := aH.reader.GetTopLevelOperations(r.Context(), start, end, services)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) dependencyGraph(w http.ResponseWriter, r *http.Request) {
query, err := parseGetServicesRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, err := aH.reader.GetDependencyGraph(r.Context(), query)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getServicesList(w http.ResponseWriter, r *http.Request) {
result, err := aH.reader.GetServicesList(r.Context())
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) SearchTraces(w http.ResponseWriter, r *http.Request) {
params, err := ParseSearchTracesParams(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, "Error reading params")
return
}
result, err := aH.reader.SearchTraces(r.Context(), params)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) listErrors(w http.ResponseWriter, r *http.Request) {
query, err := parseListErrorsRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, apiErr := aH.reader.ListErrors(r.Context(), query)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) countErrors(w http.ResponseWriter, r *http.Request) {
query, err := parseCountErrorsRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, apiErr := aH.reader.CountErrors(r.Context(), query)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getErrorFromErrorID(w http.ResponseWriter, r *http.Request) {
query, err := parseGetErrorRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, apiErr := aH.reader.GetErrorFromErrorID(r.Context(), query)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getNextPrevErrorIDs(w http.ResponseWriter, r *http.Request) {
query, err := parseGetErrorRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, apiErr := aH.reader.GetNextPrevErrorIDs(r.Context(), query)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getErrorFromGroupID(w http.ResponseWriter, r *http.Request) {
query, err := parseGetErrorRequest(r)
if aH.HandleError(w, err, http.StatusBadRequest) {
return
}
result, apiErr := aH.reader.GetErrorFromGroupID(r.Context(), query)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) setCustomRetentionTTL(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, errv2 := authtypes.ClaimsFromContext(ctx)
if errv2 != nil {
render.Error(w, errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get org id from context"))
return
}
var params retentiontypes.CustomRetentionTTLParams
if err := json.NewDecoder(r.Body).Decode(&params); err != nil {
render.Error(w, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "Invalid data"))
return
}
// Context is not used here as TTL is long duration DB operation
result, apiErr := aH.reader.SetTTLV2(context.Background(), claims.OrgID, &params)
if apiErr != nil {
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInternal, apiErr.Error()))
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getCustomRetentionTTL(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, errv2 := authtypes.ClaimsFromContext(ctx)
if errv2 != nil {
render.Error(w, errors.New(errors.TypeInternal, errors.CodeInternal, "failed to get org id from context"))
return
}
result, apiErr := aH.reader.GetCustomRetentionTTL(r.Context(), claims.OrgID)
if apiErr != nil {
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInternal, apiErr.Error()))
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getDisks(w http.ResponseWriter, r *http.Request) {
result, apiErr := aH.reader.GetDisks(context.Background())
if apiErr != nil && aH.HandleError(w, apiErr.Err, http.StatusInternalServerError) {
return
}
aH.WriteJSON(w, r, result)
}
func (aH *APIHandler) getVersion(w http.ResponseWriter, r *http.Request) {
// EE is "N" and the constant is the point: this build has NO enterprise edition.
// It reported "Y" while the EE features were compiled out, so the frontend took
// the enterprise bootstrap path and called SigNoz's Zeus licensing service, which
// 404s here — a flag claiming a capability the binary does not have.
//
// The field STAYS rather than being deleted, and stays a single expression, so
// enabling an edition later is one edit in one place. When there is something to
// gate, this reads Hanzo's OWN entitlement (GET /v1/entitlements on api.hanzo.ai,
// cloud/apps/entitlements — live and auth-gated today, 403 not 404) — never Zeus,
// which is SigNoz's commercial licensing service and not ours to call.
versionResponse := model.GetVersionResponse{
Version: version.Info.Version(),
EE: "N",
SetupCompleted: aH.SetupCompleted,
}
aH.WriteJSON(w, r, versionResponse)
}
// getHealth is used to check the health of the service.
// 'live' query param can be used to check liveliness of
// the service by checking the database connection.
func (aH *APIHandler) getHealth(w http.ResponseWriter, r *http.Request) {
_, ok := r.URL.Query()["live"]
if ok {
err := aH.reader.CheckDatastore(r.Context())
if err != nil {
RespondError(w, &model.ApiError{Err: err, Typ: model.ErrorStatusServiceUnavailable}, nil)
return
}
}
aH.WriteJSON(w, r, map[string]string{"status": "ok"})
}
func (aH *APIHandler) HandleError(w http.ResponseWriter, err error, statusCode int) bool {
if err == nil {
return false
}
if statusCode == http.StatusInternalServerError {
aH.logger.Error("internal server error in http handler", errors.Attr(err))
}
structuredResp := structuredResponse{
Errors: []structuredError{
{
Code: statusCode,
Msg: err.Error(),
},
},
}
resp, _ := json.Marshal(&structuredResp)
http.Error(w, string(resp), statusCode)
return true
}
func (aH *APIHandler) WriteJSON(w http.ResponseWriter, r *http.Request, response interface{}) {
marshall := json.Marshal
if prettyPrint := r.FormValue("pretty"); prettyPrint != "" && prettyPrint != "false" {
marshall = func(v interface{}) ([]byte, error) {
return json.MarshalIndent(v, "", " ")
}
}
resp, _ := marshall(response)
w.Header().Set("Content-Type", "application/json")
w.Write(resp)
}
// not using md5 hashing as the plain string would work
func uniqueIdentifier(params []string, separator string) string {
return strings.Join(params, separator)
}
func (aH *APIHandler) onboardProducers(
w http.ResponseWriter, r *http.Request,
) {
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
chq, err := kafka.BuildDatastoreQuery(messagingQueue, kafka.KafkaQueue, "onboard_producers")
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build datastore query for onboard producers", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, err)
return
}
var entries []kafka.OnboardingResponse
for _, result := range results {
for key, value := range result.Data {
var message, attribute, status string
intValue := int(*value.(*uint8))
if key == "entries" {
attribute = "telemetry ingestion"
if intValue != 0 {
entries = nil
entry := kafka.OnboardingResponse{
Attribute: attribute,
Message: "No data available in the given time range",
Status: "0",
}
entries = append(entries, entry)
break
} else {
status = "1"
}
} else if key == "queue" {
attribute = "messaging.system"
if intValue != 0 {
status = "0"
message = "messaging.system attribute is not present or not equal to kafka in your spans"
} else {
status = "1"
}
} else if key == "kind" {
attribute = "kind"
if intValue != 0 {
status = "0"
message = "check if your producer spans has kind=4 as attribute"
} else {
status = "1"
}
} else if key == "destination" {
attribute = "messaging.destination.name"
if intValue != 0 {
status = "0"
message = "messaging.destination.name attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "partition" {
attribute = "messaging.destination.partition.id"
if intValue != 0 {
status = "0"
message = "messaging.destination.partition.id attribute is not present in your spans"
} else {
status = "1"
}
}
entry := kafka.OnboardingResponse{
Attribute: attribute,
Message: message,
Status: status,
}
entries = append(entries, entry)
}
}
sort.Slice(entries, func(i, j int) bool {
return entries[i].Attribute < entries[j].Attribute
})
aH.Respond(w, entries)
}
func (aH *APIHandler) onboardConsumers(
w http.ResponseWriter, r *http.Request,
) {
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
chq, err := kafka.BuildDatastoreQuery(messagingQueue, kafka.KafkaQueue, "onboard_consumers")
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build datastore query for onboard consumers", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
result, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, err)
return
}
var entries []kafka.OnboardingResponse
for _, result := range result {
for key, value := range result.Data {
var message, attribute, status string
intValue := int(*value.(*uint8))
if key == "entries" {
attribute = "telemetry ingestion"
if intValue != 0 {
entries = nil
entry := kafka.OnboardingResponse{
Attribute: attribute,
Message: "No data available in the given time range",
Status: "0",
}
entries = append(entries, entry)
break
} else {
status = "1"
}
} else if key == "queue" {
attribute = "messaging.system"
if intValue != 0 {
status = "0"
message = "messaging.system attribute is not present or not equal to kafka in your spans"
} else {
status = "1"
}
} else if key == "kind" {
attribute = "kind"
if intValue != 0 {
status = "0"
message = "check if your consumer spans has kind=5 as attribute"
} else {
status = "1"
}
} else if key == "destination" {
attribute = "messaging.destination.name"
if intValue != 0 {
status = "0"
message = "messaging.destination.name attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "partition" {
attribute = "messaging.destination.partition.id"
if intValue != 0 {
status = "0"
message = "messaging.destination.partition.id attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "svc" {
attribute = "service_name"
if intValue != 0 {
status = "0"
message = "service_name attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "cgroup" {
attribute = "messaging.kafka.consumer.group"
if intValue != 0 {
status = "0"
message = "messaging.kafka.consumer.group attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "bodysize" {
attribute = "messaging.message.body.size"
if intValue != 0 {
status = "0"
message = "messaging.message.body.size attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "clientid" {
attribute = "messaging.client_id"
if intValue != 0 {
status = "0"
message = "messaging.client_id attribute is not present in your spans"
} else {
status = "1"
}
} else if key == "instanceid" {
attribute = "service.instance.id"
if intValue != 0 {
status = "0"
message = "service.instance.id attribute is not present in your spans"
} else {
status = "1"
}
}
entry := kafka.OnboardingResponse{
Attribute: attribute,
Message: message,
Status: status,
}
entries = append(entries, entry)
}
}
sort.Slice(entries, func(i, j int) bool {
return entries[i].Attribute < entries[j].Attribute
})
aH.Respond(w, entries)
}
func (aH *APIHandler) onboardKafka(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
queryRangeParams, err := kafka.BuildBuilderQueriesKafkaOnboarding(messagingQueue)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build kafka onboarding queries", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "onboardKafka",
})
results, errQueriesByName, err := aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
var entries []kafka.OnboardingResponse
var fetchLatencyState, consumerLagState bool
for _, result := range results {
for _, series := range result.Series {
for _, point := range series.Points {
pointValue := point.Value
if pointValue > 0 {
if result.QueryName == "fetch_latency" {
fetchLatencyState = true
break
}
if result.QueryName == "consumer_lag" {
consumerLagState = true
break
}
}
}
}
}
var kafkaConsumerFetchLatencyAvg string = "kafka_consumer_fetch_latency_avg"
var kafkaConsumerLag string = "kafka_consumer_group_lag"
if constants.IsDotMetricsEnabled {
kafkaConsumerLag = "kafka.consumer_group.lag"
kafkaConsumerFetchLatencyAvg = "kafka.consumer.fetch_latency_avg"
}
if !fetchLatencyState && !consumerLagState {
entries = append(entries, kafka.OnboardingResponse{
Attribute: "telemetry ingestion",
Message: "No data available in the given time range",
Status: "0",
})
}
if !fetchLatencyState {
entries = append(entries, kafka.OnboardingResponse{
Attribute: kafkaConsumerFetchLatencyAvg,
Message: "Metric kafka_consumer_fetch_latency_avg is not present in the given time range.",
Status: "0",
})
} else {
entries = append(entries, kafka.OnboardingResponse{
Attribute: kafkaConsumerFetchLatencyAvg,
Status: "1",
})
}
if !consumerLagState {
entries = append(entries, kafka.OnboardingResponse{
Attribute: kafkaConsumerLag,
Message: "Metric kafka_consumer_group_lag is not present in the given time range.",
Status: "0",
})
} else {
entries = append(entries, kafka.OnboardingResponse{
Attribute: kafkaConsumerLag,
Status: "1",
})
}
aH.Respond(w, entries)
}
func (aH *APIHandler) getNetworkData(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
attributeCache := &kafka.Clients{
Hash: make(map[string]struct{}),
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
queryRangeParams, err := kafka.BuildQRParamsWithCache(messagingQueue, "throughput", attributeCache)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for throughput", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getNetworkData",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
for _, res := range result {
for _, series := range res.Series {
clientID, clientIDOk := series.Labels["client_id"]
serviceInstanceID, serviceInstanceIDOk := series.Labels["service_instance_id"]
serviceName, serviceNameOk := series.Labels["service_name"]
params := []string{clientID, serviceInstanceID, serviceName}
hashKey := uniqueIdentifier(params, "#")
_, ok := attributeCache.Hash[hashKey]
if clientIDOk && serviceInstanceIDOk && serviceNameOk && !ok {
attributeCache.Hash[hashKey] = struct{}{}
attributeCache.ClientID = append(attributeCache.ClientID, clientID)
attributeCache.ServiceInstanceID = append(attributeCache.ServiceInstanceID, serviceInstanceID)
attributeCache.ServiceName = append(attributeCache.ServiceName, serviceName)
}
}
}
queryRangeParams, err = kafka.BuildQRParamsWithCache(messagingQueue, "fetch-latency", attributeCache)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for fetch latency", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for fetch latency", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
resultFetchLatency, errQueriesByNameFetchLatency, err := aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByNameFetchLatency)
return
}
latencyColumn := &v3.Result{QueryName: "latency"}
var latencySeries []*v3.Series
for _, res := range resultFetchLatency {
for _, series := range res.Series {
clientID, clientIDOk := series.Labels["client_id"]
serviceInstanceID, serviceInstanceIDOk := series.Labels["service_instance_id"]
serviceName, serviceNameOk := series.Labels["service_name"]
params := []string{clientID, serviceInstanceID, serviceName}
hashKey := uniqueIdentifier(params, "#")
_, ok := attributeCache.Hash[hashKey]
if clientIDOk && serviceInstanceIDOk && serviceNameOk && ok {
latencySeries = append(latencySeries, series)
}
}
}
latencyColumn.Series = latencySeries
result = append(result, latencyColumn)
resultFetchLatency = postprocess.TransformToTableForBuilderQueries(result, queryRangeParams)
resp := v3.QueryRangeResponse{
Result: resultFetchLatency,
}
aH.Respond(w, resp)
}
func (aH *APIHandler) getProducerData(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
// parse the query params to retrieve the messaging queue struct
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getProducerData",
})
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(ctx, flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "producer", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for producer", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for producer", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
func (aH *APIHandler) getConsumerData(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "consumer", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for consumer", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for consumer", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getConsumerData",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s1
func (aH *APIHandler) getPartitionOverviewLatencyData(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "producer-topic-throughput", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for producer topic throughput", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for producer topic throughput", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getPartitionOverviewLatencyData",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s1
func (aH *APIHandler) getConsumerPartitionLatencyData(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "consumer_partition_latency", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for consumer partition latency", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for consumer partition latency", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getConsumerPartitionLatencyData",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s3 p overview
// fetch traces
// cache attributes
// fetch byte rate metrics
func (aH *APIHandler) getProducerThroughputOverview(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
attributeCache := &kafka.Clients{
Hash: make(map[string]struct{}),
}
producerQueryRangeParams, err := kafka.BuildQRParamsWithCache(messagingQueue, "producer-throughput-overview", attributeCache)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for producer throughput overview", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(producerQueryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for producer throughput overview", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getProducerThroughputOverview",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, producerQueryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
for _, res := range result {
for _, series := range res.Series {
serviceName, serviceNameOk := series.Labels["service_name"]
topicName, topicNameOk := series.Labels["topic"]
params := []string{serviceName, topicName}
hashKey := uniqueIdentifier(params, "#")
_, ok := attributeCache.Hash[hashKey]
if topicNameOk && serviceNameOk && !ok {
attributeCache.Hash[hashKey] = struct{}{}
attributeCache.TopicName = append(attributeCache.TopicName, topicName)
attributeCache.ServiceName = append(attributeCache.ServiceName, serviceName)
}
}
}
queryRangeParams, err := kafka.BuildQRParamsWithCache(messagingQueue, "producer-throughput-overview-byte-rate", attributeCache)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for producer throughput byte rate", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for producer throughput byte rate", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
resultFetchLatency, errQueriesByNameFetchLatency, err := aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByNameFetchLatency)
return
}
byteRateColumn := &v3.Result{QueryName: "byte_rate"}
var byteRateSeries []*v3.Series
for _, res := range resultFetchLatency {
for _, series := range res.Series {
topic, topicOk := series.Labels["topic"]
serviceName, serviceNameOk := series.Labels["service_name"]
params := []string{serviceName, topic}
hashKey := uniqueIdentifier(params, "#")
_, ok := attributeCache.Hash[hashKey]
if topicOk && serviceNameOk && ok {
byteRateSeries = append(byteRateSeries, series)
}
}
}
byteRateColumn.Series = byteRateSeries
var latencyColumnResult []*v3.Result
latencyColumnResult = append(latencyColumnResult, byteRateColumn)
resultFetchLatency = postprocess.TransformToTableForBuilderQueries(latencyColumnResult, queryRangeParams)
result = postprocess.TransformToTableForDatastoreQueries(result)
result = append(result, resultFetchLatency[0])
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s3 p details
func (aH *APIHandler) getProducerThroughputDetails(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "producer-throughput-details", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for producer throughput details", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for producer throughput details", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getProducerThroughputDetails",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s3 c overview
func (aH *APIHandler) getConsumerThroughputOverview(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "consumer-throughput-overview", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for consumer throughput overview", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for consumer throughput overview", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getConsumerThroughputOverview",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s3 c details
func (aH *APIHandler) getConsumerThroughputDetails(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "consumer-throughput-details", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for consumer throughput details", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for consumer throughput details", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getConsumerThroughputDetails",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
result = postprocess.TransformToTableForDatastoreQueries(result)
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
// s4
// needs logic to parse duration
// needs logic to get the percentage
// show 10 traces
func (aH *APIHandler) getProducerConsumerEval(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
messagingQueue, apiErr := ParseKafkaQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse kafka queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
kafkaSpanEval := aH.O11y.Flagger.BooleanOrEmpty(r.Context(), flagger.FeatureKafkaSpanEval, evalCtx)
queryRangeParams, err := kafka.BuildQueryRangeParams(messagingQueue, "producer-consumer-eval", kafkaSpanEval)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build query range params for producer consumer eval", errors.Attr(err))
RespondError(w, &model.ApiError{
Typ: model.ErrorBadData,
Err: err,
}, nil)
return
}
if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
aH.logger.ErrorContext(r.Context(), "failed to validate query range params for producer consumer eval", errors.Attr(err))
RespondError(w, apiErr, nil)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getProducerConsumerEval",
})
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
func (aH *APIHandler) ListIntegrations(
w http.ResponseWriter, r *http.Request,
) {
params := map[string]string{}
for k, values := range r.URL.Query() {
params[k] = values[0]
}
claims, errv2 := authtypes.ClaimsFromContext(r.Context())
if errv2 != nil {
render.Error(w, errv2)
return
}
resp, apiErr := aH.IntegrationsController.ListIntegrations(
r.Context(), claims.OrgID, params,
)
if apiErr != nil {
RespondError(w, apiErr, "Failed to fetch integrations")
return
}
aH.Respond(w, resp)
}
func (aH *APIHandler) GetIntegration(
w http.ResponseWriter, r *http.Request,
) {
integrationId := coretypes.Param(r, "integrationId")
claims, errv2 := authtypes.ClaimsFromContext(r.Context())
if errv2 != nil {
render.Error(w, errv2)
return
}
integration, apiErr := aH.IntegrationsController.GetIntegration(
r.Context(), claims.OrgID, integrationId,
)
if apiErr != nil {
RespondError(w, apiErr, "Failed to fetch integration details")
return
}
aH.Respond(w, integration)
}
func (aH *APIHandler) GetIntegrationConnectionStatus(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
integrationId := coretypes.Param(r, "integrationId")
claims, errv2 := authtypes.ClaimsFromContext(r.Context())
if errv2 != nil {
render.Error(w, errv2)
return
}
isInstalled, apiErr := aH.IntegrationsController.IsIntegrationInstalled(
r.Context(), claims.OrgID, integrationId,
)
if apiErr != nil {
RespondError(w, apiErr, "failed to check if integration is installed")
return
}
// Do not spend resources calculating connection status unless installed.
if !isInstalled {
aH.Respond(w, &integrations.IntegrationConnectionStatus{})
return
}
connectionTests, apiErr := aH.IntegrationsController.GetIntegrationConnectionTests(
r.Context(), claims.OrgID, integrationId,
)
if apiErr != nil {
RespondError(w, apiErr, "failed to fetch integration connection tests")
return
}
lookbackSecondsStr := r.URL.Query().Get("lookback_seconds")
lookbackSeconds, err := strconv.ParseInt(lookbackSecondsStr, 10, 64)
if err != nil {
lookbackSeconds = 15 * 60
}
connectionStatus, apiErr := aH.calculateConnectionStatus(
r.Context(), orgID, connectionTests, lookbackSeconds,
)
if apiErr != nil {
RespondError(w, apiErr, "Failed to calculate integration connection status")
return
}
aH.Respond(w, connectionStatus)
}
func (aH *APIHandler) calculateConnectionStatus(
ctx context.Context,
orgID valuer.UUID,
connectionTests *integrations.IntegrationConnectionTests,
lookbackSeconds int64,
) (*integrations.IntegrationConnectionStatus, *model.ApiError) {
// Calculate connection status for signals in parallel
result := &integrations.IntegrationConnectionStatus{}
errors := []*model.ApiError{}
var resultLock sync.Mutex
var wg sync.WaitGroup
// Calculate logs connection status
wg.Add(1)
go func() {
defer wg.Done()
logsConnStatus, apiErr := aH.calculateLogsConnectionStatus(ctx, orgID, connectionTests.Logs, lookbackSeconds)
resultLock.Lock()
defer resultLock.Unlock()
if apiErr != nil {
errors = append(errors, apiErr)
} else {
result.Logs = logsConnStatus
}
}()
// Calculate metrics connection status
wg.Add(1)
go func() {
defer wg.Done()
if len(connectionTests.Metrics) < 1 {
return
}
statusForLastReceivedMetric, apiErr := aH.reader.GetLatestReceivedMetric(
ctx, orgID, connectionTests.Metrics, nil,
)
resultLock.Lock()
defer resultLock.Unlock()
if apiErr != nil {
errors = append(errors, apiErr)
} else if statusForLastReceivedMetric != nil {
resourceSummaryParts := []string{}
for k, v := range statusForLastReceivedMetric.LastReceivedLabels {
interestingLabels := []string{
"container_name", "host_name", "node_name",
"pod_name", "deployment_name", "cluster_name",
"namespace_name", "job_name", "service_name",
}
isInterestingKey := !strings.HasPrefix(k, "_") && slices.ContainsFunc(
interestingLabels, func(l string) bool { return strings.Contains(k, l) },
)
if isInterestingKey {
resourceSummaryParts = append(resourceSummaryParts, fmt.Sprintf(
"%s=%s", k, v,
))
}
}
result.Metrics = &integrations.SignalConnectionStatus{
LastReceivedFrom: strings.Join(resourceSummaryParts, ", "),
LastReceivedTsMillis: statusForLastReceivedMetric.LastReceivedTsMillis,
}
}
}()
wg.Wait()
if len(errors) > 0 {
return nil, errors[0]
}
return result, nil
}
func (aH *APIHandler) calculateLogsConnectionStatus(ctx context.Context, orgID valuer.UUID, logsConnectionTest *integrations.LogsConnectionTest, lookbackSeconds int64) (*integrations.SignalConnectionStatus, *model.ApiError) {
if logsConnectionTest == nil {
return nil, nil
}
logsConnTestFilter := &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: logsConnectionTest.AttributeKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeTag,
},
Operator: "=",
Value: logsConnectionTest.AttributeValue,
},
},
}
qrParams := &v3.QueryRangeParamsV3{
Start: time.Now().UnixMilli() - (lookbackSeconds * 1000),
End: time.Now().UnixMilli(),
CompositeQuery: &v3.CompositeQuery{
PanelType: v3.PanelTypeList,
QueryType: v3.QueryTypeBuilder,
BuilderQueries: map[string]*v3.BuilderQuery{
"A": {
PageSize: 1,
Filters: logsConnTestFilter,
QueryName: "A",
DataSource: v3.DataSourceLogs,
Expression: "A",
AggregateOperator: v3.AggregateOperatorNoOp,
},
},
},
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "calculateLogsConnectionStatus",
})
queryRes, _, err := aH.querier.QueryRange(ctx, orgID, qrParams)
if err != nil {
return nil, model.InternalError(fmt.Errorf(
"could not query for integration connection status: %w", err,
))
}
if len(queryRes) > 0 && queryRes[0].List != nil && len(queryRes[0].List) > 0 {
lastLog := queryRes[0].List[0]
resourceSummaryParts := []string{}
lastLogResourceAttribs := lastLog.Data["resources_string"]
if lastLogResourceAttribs != nil {
resourceAttribs, ok := lastLogResourceAttribs.(*map[string]string)
if !ok {
return nil, model.InternalError(fmt.Errorf(
"could not cast log resource attribs",
))
}
for k, v := range *resourceAttribs {
resourceSummaryParts = append(resourceSummaryParts, fmt.Sprintf(
"%s=%s", k, v,
))
}
}
lastLogResourceSummary := strings.Join(resourceSummaryParts, ", ")
return &integrations.SignalConnectionStatus{
LastReceivedTsMillis: lastLog.Timestamp.UnixMilli(),
LastReceivedFrom: lastLogResourceSummary,
}, nil
}
return nil, nil
}
func (aH *APIHandler) InstallIntegration(w http.ResponseWriter, r *http.Request) {
req := integrations.InstallIntegrationRequest{}
err := json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, model.BadRequest(err), nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
integration, apiErr := aH.IntegrationsController.Install(
r.Context(), claims.OrgID, &req, claims.Email, valuer.MustNewUUID(claims.IdentityID()),
)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.Respond(w, integration)
}
func (aH *APIHandler) UninstallIntegration(w http.ResponseWriter, r *http.Request) {
req := integrations.UninstallIntegrationRequest{}
err := json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, model.BadRequest(err), nil)
return
}
claims, errv2 := authtypes.ClaimsFromContext(r.Context())
if errv2 != nil {
render.Error(w, errv2)
return
}
apiErr := aH.IntegrationsController.Uninstall(r.Context(), claims.OrgID, &req)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.Respond(w, map[string]interface{}{})
}
func (aH *APIHandler) logFields(w http.ResponseWriter, r *http.Request) {
fields, apiErr := aH.reader.GetLogFields(r.Context())
if apiErr != nil {
RespondError(w, apiErr, "Failed to fetch fields from the DB")
return
}
aH.WriteJSON(w, r, fields)
}
func (aH *APIHandler) logFieldUpdate(w http.ResponseWriter, r *http.Request) {
field := model.UpdateField{}
if err := json.NewDecoder(r.Body).Decode(&field); err != nil {
apiErr := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErr, "Failed to decode payload")
return
}
err := logs.ValidateUpdateFieldPayload(&field)
if err != nil {
apiErr := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErr, "Incorrect payload")
return
}
apiErr := aH.reader.UpdateLogField(r.Context(), &field)
if apiErr != nil {
RespondError(w, apiErr, "Failed to update field in the DB")
return
}
aH.WriteJSON(w, r, field)
}
// getLogs serves GET /v1/o11y/logs — the most recent logs over the
// query window, newest first. Reads the configured logs table via the reader
// (real Datastore rows), replacing the former {"results":[]} stub. Params:
// limit (default 100, max 1000), timestampStart / timestampEnd (nanosecond
// epochs; default the last 15 minutes). The read is the caller's tenant only;
// a request that carries none is 403. A read error surfaces as an error, never
// a fabricated empty list.
func (aH *APIHandler) getLogs(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
limit := 100
if v, err := strconv.Atoi(strings.TrimSpace(q.Get("limit"))); err == nil && v > 0 {
limit = v
}
end := time.Now().UnixNano()
if v, err := strconv.ParseInt(strings.TrimSpace(q.Get("timestampEnd")), 10, 64); err == nil && v > 0 {
end = v
}
start := end - int64(15*time.Minute)
if v, err := strconv.ParseInt(strings.TrimSpace(q.Get("timestampStart")), 10, 64); err == nil && v > 0 {
start = v
}
if _, err := authtypes.TenantFromContext(r.Context()); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorForbidden, Err: err}, nil)
return
}
rows, err := aH.reader.GetRecentLogs(r.Context(), start, end, limit)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, "Failed to fetch logs from the DB")
return
}
results := make([]map[string]interface{}, 0, len(rows))
for _, row := range rows {
item := map[string]interface{}{"timestamp": row.Timestamp.UnixNano()}
for k, v := range row.Data {
item[k] = v
}
results = append(results, item)
}
aH.WriteJSON(w, r, map[string]interface{}{"results": results})
}
func (aH *APIHandler) logAggregate(w http.ResponseWriter, r *http.Request) {
aH.WriteJSON(w, r, model.GetLogsAggregatesResponse{})
}
func parseAgentConfigVersion(r *http.Request) (int, error) {
versionString := coretypes.Param(r, "version")
if versionString == "latest" {
return -1, nil
}
version64, err := strconv.ParseInt(versionString, 0, 8)
if err != nil {
return 0, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "invalid version number")
}
if version64 <= 0 {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid version number")
}
return int(version64), nil
}
func (aH *APIHandler) PreviewLogsPipelinesHandler(w http.ResponseWriter, r *http.Request) {
req := logparsingpipeline.PipelinesPreviewRequest{}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
render.Error(w, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "failed to decode request body"))
return
}
resultLogs, err := aH.LogsParsingPipelineController.PreviewLogsPipelines(r.Context(), &req)
if err != nil {
render.Error(w, err)
return
}
aH.Respond(w, resultLogs)
}
func (aH *APIHandler) ListLogsPipelinesHandler(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, errv2 := valuer.NewUUID(claims.OrgID)
if errv2 != nil {
render.Error(w, errv2)
return
}
version, err := parseAgentConfigVersion(r)
if err != nil {
render.Error(w, err)
return
}
var payload *logparsingpipeline.PipelinesResponse
if version != -1 {
payload, err = aH.listLogsPipelinesByVersion(r.Context(), orgID, version)
} else {
payload, err = aH.listLogsPipelines(r.Context(), orgID)
}
if err != nil {
render.Error(w, err)
return
}
aH.Respond(w, payload)
}
// listLogsPipelines lists logs pipelines for latest version
func (aH *APIHandler) listLogsPipelines(ctx context.Context, orgID valuer.UUID) (
*logparsingpipeline.PipelinesResponse, error,
) {
// get latest agent config
latestVersion := -1
lastestConfig, err := agentConf.GetLatestVersion(ctx, orgID, opamptypes.ElementTypeLogPipelines)
if err != nil && !errors.Ast(err, errors.TypeNotFound) {
return nil, err
}
if lastestConfig != nil {
latestVersion = lastestConfig.Version
}
payload, err := aH.LogsParsingPipelineController.GetPipelinesByVersion(ctx, orgID, latestVersion)
if err != nil {
return nil, err
}
// todo(Nitya): make a new API for history pagination
limit := 10
history, err := agentConf.GetConfigHistory(ctx, orgID, opamptypes.ElementTypeLogPipelines, limit)
if err != nil {
return nil, err
}
payload.History = history
return payload, nil
}
// listLogsPipelinesByVersion lists pipelines along with config version history
func (aH *APIHandler) listLogsPipelinesByVersion(ctx context.Context, orgID valuer.UUID, version int) (
*logparsingpipeline.PipelinesResponse, error,
) {
payload, err := aH.LogsParsingPipelineController.GetPipelinesByVersion(ctx, orgID, version)
if err != nil {
return nil, err
}
// todo(Nitya): make a new API for history pagination
limit := 10
history, err := agentConf.GetConfigHistory(ctx, orgID, opamptypes.ElementTypeLogPipelines, limit)
if err != nil {
return nil, err
}
payload.History = history
return payload, nil
}
func (aH *APIHandler) CreateLogsPipeline(w http.ResponseWriter, r *http.Request) {
claims, errv2 := authtypes.ClaimsFromContext(r.Context())
if errv2 != nil {
render.Error(w, errv2)
return
}
// prepare config by calling gen func
orgID, errv2 := valuer.NewUUID(claims.OrgID)
if errv2 != nil {
render.Error(w, errv2)
return
}
userID, errv2 := valuer.NewUUID(claims.UserID)
if errv2 != nil {
render.Error(w, errv2)
return
}
req := pipelinetypes.PostablePipelines{}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, model.BadRequest(err), nil)
return
}
createPipeline := func(
ctx context.Context,
postable []pipelinetypes.PostablePipeline,
) (*logparsingpipeline.PipelinesResponse, error) {
if len(postable) == 0 {
aH.logger.WarnContext(r.Context(), "found no pipelines in the http request, this will delete all the pipelines")
}
err := aH.LogsParsingPipelineController.ValidatePipelines(ctx, postable)
if err != nil {
return nil, err
}
return aH.LogsParsingPipelineController.ApplyPipelines(ctx, orgID, userID, postable)
}
res, err := createPipeline(r.Context(), req.Pipelines)
if err != nil {
render.Error(w, err)
return
}
aH.Respond(w, res)
}
func (aH *APIHandler) autocompleteAggregateAttributes(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
var response *v3.AggregateAttributeResponse
req, err := parseAggregateAttributeRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
switch req.DataSource {
case v3.DataSourceMetrics:
response, err = aH.reader.GetMetricAggregateAttributes(r.Context(), orgID, req, false)
case v3.DataSourceLogs:
response, err = aH.reader.GetLogAggregateAttributes(r.Context(), req)
case v3.DataSourceTraces:
response, err = aH.reader.GetTraceAggregateAttributes(r.Context(), req)
case v3.DataSourceMeter:
response, err = aH.reader.GetMeterAggregateAttributes(r.Context(), orgID, req)
default:
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid data source")}, nil)
return
}
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
aH.Respond(w, response)
}
func (aH *APIHandler) getQueryBuilderSuggestions(w http.ResponseWriter, r *http.Request) {
req, err := parseQBFilterSuggestionsRequest(r)
if err != nil {
RespondError(w, err, nil)
return
}
if req.DataSource != v3.DataSourceLogs {
// Support for traces and metrics might come later
RespondError(w, model.BadRequest(
fmt.Errorf("suggestions not supported for %s", req.DataSource),
), nil)
return
}
response, apiErr := aH.reader.GetQBFilterSuggestionsForLogs(r.Context(), req)
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
aH.Respond(w, response)
}
func (aH *APIHandler) autoCompleteAttributeKeys(w http.ResponseWriter, r *http.Request) {
var response *v3.FilterAttributeKeyResponse
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
aH.HandleError(w, err, http.StatusInternalServerError)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
switch req.DataSource {
case v3.DataSourceMetrics:
response, err = aH.reader.GetMetricAttributeKeys(r.Context(), orgID, req)
case v3.DataSourceMeter:
response, err = aH.reader.GetMeterAttributeKeys(r.Context(), req)
case v3.DataSourceLogs:
response, err = aH.reader.GetLogAttributeKeys(r.Context(), req)
case v3.DataSourceTraces:
response, err = aH.reader.GetTraceAttributeKeys(r.Context(), req)
default:
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid data source")}, nil)
return
}
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
aH.Respond(w, response)
}
func (aH *APIHandler) autoCompleteAttributeValues(w http.ResponseWriter, r *http.Request) {
var response *v3.FilterAttributeValueResponse
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
aH.HandleError(w, err, http.StatusInternalServerError)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
switch req.DataSource {
case v3.DataSourceMetrics:
response, err = aH.reader.GetMetricAttributeValues(r.Context(), orgID, req)
case v3.DataSourceLogs:
response, err = aH.reader.GetLogAttributeValues(r.Context(), req)
case v3.DataSourceTraces:
response, err = aH.reader.GetTraceAttributeValues(r.Context(), req)
default:
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid data source")}, nil)
return
}
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
aH.Respond(w, response)
}
func (aH *APIHandler) autoCompleteAttributeValuesPost(w http.ResponseWriter, r *http.Request) {
var response *v3.FilterAttributeValueResponse
req, err := parseFilterAttributeValueRequestBody(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
aH.HandleError(w, err, http.StatusInternalServerError)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
switch req.DataSource {
case v3.DataSourceMetrics:
response, err = aH.reader.GetMetricAttributeValues(r.Context(), orgID, req)
case v3.DataSourceLogs:
response, err = aH.reader.GetLogAttributeValues(r.Context(), req)
case v3.DataSourceTraces:
response, err = aH.reader.GetTraceAttributeValues(r.Context(), req)
default:
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid data source")}, nil)
return
}
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}
aH.Respond(w, response)
}
func (aH *APIHandler) getSpanKeysV3(ctx context.Context, queryRangeParams *v3.QueryRangeParamsV3) (map[string]v3.AttributeKey, error) {
data := map[string]v3.AttributeKey{}
for _, query := range queryRangeParams.CompositeQuery.BuilderQueries {
if query.DataSource == v3.DataSourceTraces {
spanKeys, err := aH.reader.GetSpanAttributeKeysByNames(ctx, logsv3.GetFieldNames(queryRangeParams.CompositeQuery))
if err != nil {
return nil, err
}
// Add timestamp as a span key to allow ordering by timestamp
spanKeys["timestamp"] = v3.AttributeKey{
Key: "timestamp",
IsColumn: true,
}
return spanKeys, nil
}
}
return data, nil
}
func (aH *APIHandler) QueryRangeV3Format(w http.ResponseWriter, r *http.Request) {
queryRangeParams, apiErrorObj := ParseQueryRangeParams(r)
if apiErrorObj != nil {
aH.logger.ErrorContext(r.Context(), "error parsing query range params", errors.Attr(apiErrorObj.Err))
RespondError(w, apiErrorObj, nil)
return
}
queryRangeParams.Version = "v3"
aH.Respond(w, queryRangeParams)
}
func (aH *APIHandler) queryRangeV3(ctx context.Context, queryRangeParams *v3.QueryRangeParamsV3, w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
var spanKeys map[string]v3.AttributeKey
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
hasLogsQuery := false
hasTracesQuery := false
for _, query := range queryRangeParams.CompositeQuery.BuilderQueries {
if query.DataSource == v3.DataSourceLogs {
hasLogsQuery = true
}
if query.DataSource == v3.DataSourceTraces {
hasTracesQuery = true
}
}
// check if any enrichment is required for logs if yes then enrich them
if logsv3.EnrichmentRequired(queryRangeParams) && hasLogsQuery {
logsFields, apiErr := aH.reader.GetLogFieldsFromNames(ctx, logsv3.GetFieldNames(queryRangeParams.CompositeQuery))
if apiErr != nil {
RespondError(w, apiErr, errQueriesByName)
return
}
// get the fields if any logs query is present
fields := model.GetLogFieldsV3(ctx, queryRangeParams, logsFields)
logsv3.Enrich(queryRangeParams, fields)
}
if hasTracesQuery {
spanKeys, err = aH.getSpanKeysV3(ctx, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorInternal, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
tracesV4.Enrich(queryRangeParams, spanKeys)
}
}
// WARN: Only works for AND operator in traces query
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
// check if traceID is used as filter (with equal/similar operator) in traces query if yes add timestamp filter to queryRange params
isUsed, traceIDs := tracesV3.TraceIdFilterUsedWithEqual(queryRangeParams)
if isUsed && len(traceIDs) > 0 {
aH.logger.DebugContext(ctx, "trace_id used as filter in traces query")
// query o11y_spans table with traceID to get min and max timestamp
min, max, err := aH.reader.GetMinAndMaxTimestampForTraceID(ctx, traceIDs)
if err == nil {
// add timestamp filter to queryRange params
tracesV3.AddTimestampFilters(min, max, queryRangeParams)
aH.logger.DebugContext(ctx, "post adding timestamp filter in traces query", "query_range_params", queryRangeParams)
}
}
}
// Hook up query progress tracking if requested
queryIdHeader := r.Header.Get("X-O11Y-QUERY-ID")
if len(queryIdHeader) > 0 {
onQueryFinished, apiErr := aH.reader.ReportQueryStartForProgressTracking(queryIdHeader)
if apiErr != nil {
aH.logger.ErrorContext(ctx, "failed to report query start for progress tracking",
"query_id", queryIdHeader, errors.Attr(apiErr),
)
} else {
// Adding queryId to the context signals datastore queries to report progress
//lint:ignore SA1029 ignore for now
ctx = context.WithValue(ctx, "queryId", queryIdHeader)
defer func() {
go onQueryFinished()
}()
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "QueryRange",
})
result, errQueriesByName, err = aH.querier.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
queryErrors := map[string]string{}
for name, err := range errQueriesByName {
queryErrors[fmt.Sprintf("Query-%s", name)] = err.Error()
}
apiErrObj := &model.ApiError{Typ: model.ErrorInternal, Err: err}
RespondError(w, apiErrObj, queryErrors)
return
}
postprocess.ApplyHavingClause(result, queryRangeParams)
postprocess.ApplyMetricLimit(result, queryRangeParams)
aH.sendQueryResultEvents(r, result, queryRangeParams, "v3")
// only adding applyFunctions instead of postProcess since experssion are
// are executed in datastore directly and we wanted to add support for timeshift
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
postprocess.ApplyFunctions(result, queryRangeParams)
}
if queryRangeParams.CompositeQuery.FillGaps {
postprocess.FillGaps(result, queryRangeParams)
}
if queryRangeParams.CompositeQuery.PanelType == v3.PanelTypeTable && queryRangeParams.FormatForWeb {
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeDatastoreSQL {
result = postprocess.TransformToTableForDatastoreQueries(result)
} else if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
result = postprocess.TransformToTableForBuilderQueries(result, queryRangeParams)
}
}
resp := v3.QueryRangeResponse{
Result: result,
}
// This checks if the time for context to complete has exceeded.
// it adds flag to notify the user of incomplete response
select {
case <-ctx.Done():
resp.ContextTimeout = true
resp.ContextTimeoutMessage = "result might contain incomplete data due to context timeout, for custom timeout set the timeout header eg:- timeout:120"
default:
break
}
aH.Respond(w, resp)
}
func (aH *APIHandler) sendQueryResultEvents(r *http.Request, result []*v3.Result, queryRangeParams *v3.QueryRangeParamsV3, version string) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
return
}
queryInfoResult := NewQueryInfoResult(queryRangeParams, version)
if !(queryInfoResult.LogsUsed || queryInfoResult.MetricsUsed || queryInfoResult.TracesUsed) {
return
}
properties := queryInfoResult.ToMap()
referrer := r.Header.Get("Referer")
if referrer == "" {
return
}
properties["referrer"] = referrer
logsExplorerMatched, _ := regexp.MatchString(`/logs/logs-explorer(?:\?.*)?$`, referrer)
traceExplorerMatched, _ := regexp.MatchString(`/traces-explorer(?:\?.*)?$`, referrer)
metricsExplorerMatched, _ := regexp.MatchString(`/metrics-explorer/explorer(?:\?.*)?$`, referrer)
dashboardMatched, _ := regexp.MatchString(`/dashboard/[a-zA-Z0-9\-]+/(new|edit)(?:\?.*)?$`, referrer)
alertMatched, _ := regexp.MatchString(`/alerts/(new|edit)(?:\?.*)?$`, referrer)
switch {
case dashboardMatched:
properties["module_name"] = "dashboard"
case alertMatched:
properties["module_name"] = "rule"
case metricsExplorerMatched:
properties["module_name"] = "metrics-explorer"
case logsExplorerMatched:
properties["module_name"] = "logs-explorer"
case traceExplorerMatched:
properties["module_name"] = "traces-explorer"
default:
return
}
if dashboardMatched {
if dashboardIDRegex, err := regexp.Compile(`/dashboard/([a-f0-9\-]+)/`); err == nil {
if matches := dashboardIDRegex.FindStringSubmatch(referrer); len(matches) > 1 {
properties["dashboard_id"] = matches[1]
}
}
if widgetIDRegex, err := regexp.Compile(`widgetId=([a-f0-9\-]+)`); err == nil {
if matches := widgetIDRegex.FindStringSubmatch(referrer); len(matches) > 1 {
properties["widget_id"] = matches[1]
}
}
}
if alertMatched {
if alertIDRegex, err := regexp.Compile(`ruleId=(\d+)`); err == nil {
if matches := alertIDRegex.FindStringSubmatch(referrer); len(matches) > 1 {
properties["rule_id"] = matches[1]
}
}
}
// Check if result is empty or has no data
if len(result) == 0 {
aH.O11y.Analytics.TrackUser(r.Context(), claims.OrgID, claims.IdentityID(), "Telemetry Query Returned Empty", properties)
return
}
// Check if first result has no series data
if len(result[0].Series) == 0 {
// Check if first result has no list data
if len(result[0].List) == 0 {
// Check if first result has no table data
if result[0].Table == nil {
aH.O11y.Analytics.TrackUser(r.Context(), claims.OrgID, claims.IdentityID(), "Telemetry Query Returned Empty", properties)
return
}
if len(result[0].Table.Rows) == 0 {
aH.O11y.Analytics.TrackUser(r.Context(), claims.OrgID, claims.IdentityID(), "Telemetry Query Returned Empty", properties)
return
}
}
}
aH.O11y.Analytics.TrackUser(r.Context(), claims.OrgID, claims.IdentityID(), "Telemetry Query Returned Results", properties)
}
func (aH *APIHandler) GetQueryProgressUpdates(w http.ResponseWriter, r *http.Request) {
// Upgrade connection to websocket, sending back the requested protocol
// value for sec-websocket-protocol
//
// Since js websocket API doesn't allow setting headers, this header is often
// used for passing auth tokens. As per websocket spec the connection will only
// succeed if the requested `Sec-Websocket-Protocol` is sent back as a header
// in the upgrade response (signifying that the protocol is supported by the server).
upgradeResponseHeaders := http.Header{}
requestedProtocol := r.Header.Get("Sec-WebSocket-Protocol")
if len(requestedProtocol) > 0 {
upgradeResponseHeaders.Add("Sec-WebSocket-Protocol", requestedProtocol)
}
c, err := aH.Upgrader.Upgrade(w, r, upgradeResponseHeaders)
if err != nil {
RespondError(w, model.InternalError(fmt.Errorf(
"couldn't upgrade connection: %w", err,
)), nil)
return
}
defer c.Close()
// Websocket upgrade complete. Subscribe to query progress and send updates to client
//
// Note: we handle any subscription problems (queryId query param missing or query already complete etc)
// after the websocket connection upgrade by closing the channel.
// The other option would be to handle the errors before websocket upgrade by sending an
// error response instead of the upgrade response, but that leads to a generic websocket
// connection failure on the client.
queryId := r.URL.Query().Get("q")
progressCh, unsubscribe, apiErr := aH.reader.SubscribeToQueryProgress(queryId)
if apiErr != nil {
// Shouldn't happen unless query progress requested after query finished
aH.logger.WarnContext(r.Context(), "failed to subscribe to query progress",
"query_id", queryId, errors.Attr(apiErr),
)
return
}
defer func() { go unsubscribe() }()
for queryProgress := range progressCh {
msg, err := json.Marshal(queryProgress)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to serialize progress message",
"query_id", queryId, "progress", queryProgress, errors.Attr(err),
)
continue
}
err = c.WriteMessage(websocket.TextMessage, msg)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to write progress message to websocket",
"query_id", queryId, "msg", string(msg), errors.Attr(err),
)
break
} else {
aH.logger.DebugContext(r.Context(), "wrote progress message to websocket",
"query_id", queryId, "msg", string(msg),
)
}
}
}
func (aH *APIHandler) getMetricMetadata(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
metricName := r.URL.Query().Get("metricName")
serviceName := r.URL.Query().Get("serviceName")
metricMetadata, err := aH.reader.GetMetricMetadata(r.Context(), orgID, metricName, serviceName)
if err != nil {
RespondError(w, &model.ApiError{Err: err, Typ: model.ErrorInternal}, nil)
return
}
aH.WriteJSON(w, r, metricMetadata)
}
func (aH *APIHandler) queryRangeV4(ctx context.Context, queryRangeParams *v3.QueryRangeParamsV3, w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
var result []*v3.Result
var errQueriesByName map[string]error
var spanKeys map[string]v3.AttributeKey
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
hasLogsQuery := false
hasTracesQuery := false
for _, query := range queryRangeParams.CompositeQuery.BuilderQueries {
if query.DataSource == v3.DataSourceLogs {
hasLogsQuery = true
}
if query.DataSource == v3.DataSourceTraces {
hasTracesQuery = true
}
}
// check if any enrichment is required for logs if yes then enrich them
if logsv3.EnrichmentRequired(queryRangeParams) && hasLogsQuery {
// get the fields if any logs query is present
logsFields, apiErr := aH.reader.GetLogFieldsFromNames(r.Context(), logsv3.GetFieldNames(queryRangeParams.CompositeQuery))
if apiErr != nil {
RespondError(w, apiErr, nil)
return
}
fields := model.GetLogFieldsV3(r.Context(), queryRangeParams, logsFields)
logsv3.Enrich(queryRangeParams, fields)
}
if hasTracesQuery {
spanKeys, err = aH.getSpanKeysV3(ctx, queryRangeParams)
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorInternal, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
tracesV4.Enrich(queryRangeParams, spanKeys)
}
}
// WARN: Only works for AND operator in traces query
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
// check if traceID is used as filter (with equal/similar operator) in traces query if yes add timestamp filter to queryRange params
isUsed, traceIDs := tracesV3.TraceIdFilterUsedWithEqual(queryRangeParams)
if isUsed && len(traceIDs) > 0 {
aH.logger.DebugContext(ctx, "trace_id used as filter in traces query")
// query o11y_spans table with traceID to get min and max timestamp
min, max, err := aH.reader.GetMinAndMaxTimestampForTraceID(ctx, traceIDs)
if err == nil {
// add timestamp filter to queryRange params
tracesV3.AddTimestampFilters(min, max, queryRangeParams)
aH.logger.DebugContext(ctx, "post adding timestamp filter in traces query", "query_range_params", queryRangeParams)
}
}
}
result, errQueriesByName, err = aH.querierV2.QueryRange(ctx, orgID, queryRangeParams)
if err != nil {
queryErrors := map[string]string{}
for name, err := range errQueriesByName {
queryErrors[fmt.Sprintf("Query-%s", name)] = err.Error()
}
apiErrObj := &model.ApiError{Typ: model.ErrorInternal, Err: err}
RespondError(w, apiErrObj, queryErrors)
return
}
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
result, err = postprocess.PostProcessResult(result, queryRangeParams)
} else if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeDatastoreSQL &&
queryRangeParams.CompositeQuery.PanelType == v3.PanelTypeTable && queryRangeParams.FormatForWeb {
result = postprocess.TransformToTableForDatastoreQueries(result)
}
if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName)
return
}
aH.sendQueryResultEvents(r, result, queryRangeParams, "v4")
resp := v3.QueryRangeResponse{
Result: result,
}
aH.Respond(w, resp)
}
func (aH *APIHandler) traceFields(w http.ResponseWriter, r *http.Request) {
fields, apiErr := aH.reader.GetTraceFields(r.Context())
if apiErr != nil {
RespondError(w, apiErr, "failed to fetch fields from the db")
return
}
aH.WriteJSON(w, r, fields)
}
func (aH *APIHandler) updateTraceField(w http.ResponseWriter, r *http.Request) {
field := model.UpdateField{}
if err := json.NewDecoder(r.Body).Decode(&field); err != nil {
apiErr := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErr, "failed to decode payload")
return
}
err := logs.ValidateUpdateFieldPayloadV2(&field)
if err != nil {
apiErr := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErr, "incorrect payload")
return
}
apiErr := aH.reader.UpdateTraceField(r.Context(), &field)
if apiErr != nil {
RespondError(w, apiErr, "failed to update field in the db")
return
}
aH.WriteJSON(w, r, field)
}
func (aH *APIHandler) getQueueOverview(w http.ResponseWriter, r *http.Request) {
queueListRequest, apiErr := ParseQueueBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse queue body", errors.Attr(apiErr.Err))
RespondError(w, apiErr, nil)
return
}
chq, err := queues2.BuildOverviewQuery(queueListRequest)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build queue overview query", errors.Attr(err))
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, model.BadRequest(err), nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) getDomainList(w http.ResponseWriter, r *http.Request) {
// Extract claims from context for organization ID
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
// Parse the request body to get third-party query parameters
thirdPartyQueryRequest, apiErr := ParseRequestBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse request body", errors.Attr(apiErr))
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, apiErr.Error()))
return
}
// Build the v5 query range request for domain listing
queryRangeRequest, err := thirdpartyapi.BuildDomainList(thirdPartyQueryRequest)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build domain list query", errors.Attr(err))
apiErrObj := errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, err.Error())
render.Error(w, apiErrObj)
return
}
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getDomainList",
})
// Execute the query using the v5 querier
result, err := aH.O11y.Querier.QueryRange(ctx, orgID, queryRangeRequest)
if err != nil {
aH.logger.ErrorContext(r.Context(), "query execution failed", errors.Attr(err))
apiErrObj := errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, err.Error())
render.Error(w, apiErrObj)
return
}
result = thirdpartyapi.FilterIntermediateColumns(result)
// Filter IP addresses if ShowIp is false
var finalResult = result
if !thirdPartyQueryRequest.ShowIp {
filteredResults := thirdpartyapi.FilterResponse([]*qbtypes.QueryRangeResponse{result})
if len(filteredResults) > 0 {
finalResult = filteredResults[0]
}
}
// Send the response
aH.Respond(w, finalResult)
}
// getDomainInfo handles requests for domain information using v5 query builder
func (aH *APIHandler) getDomainInfo(w http.ResponseWriter, r *http.Request) {
// Extract claims from context for organization ID
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
// Parse the request body to get third-party query parameters
thirdPartyQueryRequest, apiErr := ParseRequestBody(r)
if apiErr != nil {
aH.logger.ErrorContext(r.Context(), "failed to parse request body", errors.Attr(apiErr))
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, apiErr.Error()))
return
}
// Build the v5 query range request for domain info
queryRangeRequest, err := thirdpartyapi.BuildDomainInfo(thirdPartyQueryRequest)
if err != nil {
aH.logger.ErrorContext(r.Context(), "failed to build domain info query", errors.Attr(err))
apiErrObj := errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, err.Error())
render.Error(w, apiErrObj)
return
}
ctx := ctxtypes.NewContextWithCommentVals(r.Context(), map[string]string{
instrumentationtypes.CodeNamespace: "app",
instrumentationtypes.CodeFunctionName: "getDomainInfo",
})
// Execute the query using the v5 querier
result, err := aH.O11y.Querier.QueryRange(ctx, orgID, queryRangeRequest)
if err != nil {
aH.logger.ErrorContext(r.Context(), "query execution failed", errors.Attr(err))
apiErrObj := errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, err.Error())
render.Error(w, apiErrObj)
return
}
result = thirdpartyapi.FilterIntermediateColumns(result)
// Filter IP addresses if ShowIp is false
var finalResult *qbtypes.QueryRangeResponse = result
if !thirdPartyQueryRequest.ShowIp {
filteredResults := thirdpartyapi.FilterResponse([]*qbtypes.QueryRangeResponse{result})
if len(filteredResults) > 0 {
finalResult = filteredResults[0]
}
}
// Send the response
aH.Respond(w, finalResult)
}
func (aH *APIHandler) handleValidateTraces(w http.ResponseWriter, r *http.Request) {
funnelID := coretypes.Param(r, "funnel_id")
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
funnel, err := aH.O11y.Modules.TraceFunnel.Get(r.Context(), valuer.MustNewUUID(funnelID), valuer.MustNewUUID(claims.OrgID))
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("funnel not found: %v", err)}, nil)
return
}
var timeRange traceFunnels.TimeRange
if err := json.NewDecoder(r.Body).Decode(&timeRange); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding time range: %v", err)}, nil)
return
}
if len(funnel.Steps) < 2 {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("funnel must have at least 2 steps")}, nil)
return
}
chq, err := traceFunnelsModule.ValidateTraces(funnel, timeRange)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelAnalytics(w http.ResponseWriter, r *http.Request) {
funnelID := coretypes.Param(r, "funnel_id")
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
funnel, err := aH.O11y.Modules.TraceFunnel.Get(r.Context(), valuer.MustNewUUID(funnelID), valuer.MustNewUUID(claims.OrgID))
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("funnel not found: %v", err)}, nil)
return
}
var stepTransition traceFunnels.StepTransitionRequest
if err := json.NewDecoder(r.Body).Decode(&stepTransition); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding time range: %v", err)}, nil)
return
}
chq, err := traceFunnelsModule.GetFunnelAnalytics(funnel, stepTransition.TimeRange)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelStepAnalytics(w http.ResponseWriter, r *http.Request) {
funnelID := coretypes.Param(r, "funnel_id")
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
funnel, err := aH.O11y.Modules.TraceFunnel.Get(r.Context(), valuer.MustNewUUID(funnelID), valuer.MustNewUUID(claims.OrgID))
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("funnel not found: %v", err)}, nil)
return
}
var stepTransition traceFunnels.StepTransitionRequest
if err := json.NewDecoder(r.Body).Decode(&stepTransition); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding time range: %v", err)}, nil)
return
}
chq, err := traceFunnelsModule.GetFunnelStepAnalytics(funnel, stepTransition.TimeRange, stepTransition.StepStart, stepTransition.StepEnd)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleStepAnalytics(w http.ResponseWriter, r *http.Request) {
funnelID := coretypes.Param(r, "funnel_id")
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
funnel, err := aH.O11y.Modules.TraceFunnel.Get(r.Context(), valuer.MustNewUUID(funnelID), valuer.MustNewUUID(claims.OrgID))
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("funnel not found: %v", err)}, nil)
return
}
var timeRange traceFunnels.TimeRange
if err := json.NewDecoder(r.Body).Decode(&timeRange); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding time range: %v", err)}, nil)
return
}
chq, err := traceFunnelsModule.GetStepAnalytics(funnel, timeRange)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelSlowTraces(w http.ResponseWriter, r *http.Request) {
funnelID := coretypes.Param(r, "funnel_id")
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
funnel, err := aH.O11y.Modules.TraceFunnel.Get(r.Context(), valuer.MustNewUUID(funnelID), valuer.MustNewUUID(claims.OrgID))
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("funnel not found: %v", err)}, nil)
return
}
var req traceFunnels.StepTransitionRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid request body: %v", err)}, nil)
return
}
chq, err := traceFunnelsModule.GetSlowestTraces(funnel, req.TimeRange, req.StepStart, req.StepEnd)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelErrorTraces(w http.ResponseWriter, r *http.Request) {
funnelID := coretypes.Param(r, "funnel_id")
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
funnel, err := aH.O11y.Modules.TraceFunnel.Get(r.Context(), valuer.MustNewUUID(funnelID), valuer.MustNewUUID(claims.OrgID))
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorNotFound, Err: fmt.Errorf("funnel not found: %v", err)}, nil)
return
}
var req traceFunnels.StepTransitionRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid request body: %v", err)}, nil)
return
}
chq, err := traceFunnelsModule.GetErroredTraces(funnel, req.TimeRange, req.StepStart, req.StepEnd)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleValidateTracesWithPayload(w http.ResponseWriter, r *http.Request) {
var req traceFunnels.PostableFunnel
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding request: %v", err)}, nil)
return
}
if len(req.Steps) < 2 {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("funnel must have at least 2 steps")}, nil)
return
}
// Create a StorableFunnel from the request
funnel := &traceFunnels.StorableFunnel{
Steps: req.Steps,
}
chq, err := traceFunnelsModule.ValidateTraces(funnel, traceFunnels.TimeRange{
StartTime: req.StartTime,
EndTime: req.EndTime,
})
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelAnalyticsWithPayload(w http.ResponseWriter, r *http.Request) {
var req traceFunnels.PostableFunnel
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding request: %v", err)}, nil)
return
}
funnel := &traceFunnels.StorableFunnel{
Steps: req.Steps,
}
chq, err := traceFunnelsModule.GetFunnelAnalytics(funnel, traceFunnels.TimeRange{
StartTime: req.StartTime,
EndTime: req.EndTime,
})
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleStepAnalyticsWithPayload(w http.ResponseWriter, r *http.Request) {
var req traceFunnels.PostableFunnel
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding request: %v", err)}, nil)
return
}
funnel := &traceFunnels.StorableFunnel{
Steps: req.Steps,
}
chq, err := traceFunnelsModule.GetStepAnalytics(funnel, traceFunnels.TimeRange{
StartTime: req.StartTime,
EndTime: req.EndTime,
})
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelStepAnalyticsWithPayload(w http.ResponseWriter, r *http.Request) {
var req traceFunnels.PostableFunnel
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding request: %v", err)}, nil)
return
}
funnel := &traceFunnels.StorableFunnel{
Steps: req.Steps,
}
chq, err := traceFunnelsModule.GetFunnelStepAnalytics(funnel, traceFunnels.TimeRange{
StartTime: req.StartTime,
EndTime: req.EndTime,
}, req.StepStart, req.StepEnd)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelSlowTracesWithPayload(w http.ResponseWriter, r *http.Request) {
var req traceFunnels.PostableFunnel
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding request: %v", err)}, nil)
return
}
funnel := &traceFunnels.StorableFunnel{
Steps: req.Steps,
}
chq, err := traceFunnelsModule.GetSlowestTraces(funnel, traceFunnels.TimeRange{
StartTime: req.StartTime,
EndTime: req.EndTime,
}, req.StepStart, req.StepEnd)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
func (aH *APIHandler) handleFunnelErrorTracesWithPayload(w http.ResponseWriter, r *http.Request) {
var req traceFunnels.PostableFunnel
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("error decoding request: %v", err)}, nil)
return
}
funnel := &traceFunnels.StorableFunnel{
Steps: req.Steps,
}
chq, err := traceFunnelsModule.GetErroredTraces(funnel, traceFunnels.TimeRange{
StartTime: req.StartTime,
EndTime: req.EndTime,
}, req.StepStart, req.StepEnd)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building datastore query: %v", err)}, nil)
return
}
results, err := aH.reader.GetListResultV3(r.Context(), chq.Query)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error converting datastore results to list: %v", err)}, nil)
return
}
aH.Respond(w, results)
}
Reference in New Issue View Git Blame Copy Permalink