Compare commits

...

11 Commits

Author SHA1 Message Date
Nikhil Mantri
b53b95c53c Merge branch 'main' into feat/deployment_markers_v1 2026-11-03 10:54:16 +05:30
nikhilmantri0902
74198e6b29 fix(annotations): widen deployment id to 16 hex
Four bytes of the hash gave a 32-bit id space, about a 0.3% chance of two
tuples sharing an id in a 5000-record response. The id now takes 8 bytes of
the sha256. It is derived, never stored, so nothing migrates.
2026-11-02 17:19:29 +05:30
nikhilmantri0902
17f554bf02 fix(annotations): drop resources without a service name
Only version != '' was enforced, so a resource with a version but no
service.name produced a marker with an empty service. Both resource passes
now require service != '' too. Goldens updated and a no-service row added to
the integration dataset.
2026-11-02 17:16:41 +05:30
nikhilmantri0902
d6c94df486 fix(annotations): read environment through the semconv family
Both resource passes read deployment.environment only, so a resource emitting
deployment.environment.name alone got an empty environment and collapsed with
other environments of the same service and version. With resolve_semconv_families
on, the keys now come from semconv.Members and compile to the COALESCE shape the
widened filter uses; with the flag off the legacy key alone is read, keeping
extraction and filtering in lockstep. Goldens and an integration test on a
flag-on instance cover both.
2026-11-02 17:09:16 +05:30
Nikhil Mantri
532dad616f Merge branch 'main' into feat/deployment_markers_v1 2026-11-02 16:31:41 +05:30
Nikhil Mantri
ce33d9a25e Merge branch 'main' into feat/deployment_markers_v1 2026-11-02 11:30:29 +05:30
nikhilmantri0902
2f34b6f678 test(annotations): add integration tests for deployment markers
Datasets under testdata/annotations seed both signals with a baseline tuple in the
oldest bucket and markers inside the window. 01_deployments covers the response
contract, inclusive window bounds, oldest-bucket rule, signal selection, merge
winners, history edge per signal, filters, limit and validation errors. 02_authz
pins that a grant on only one signal is denied and that key-scoped grants need the
key pinned in the filter.
2026-11-01 18:10:18 +05:30
nikhilmantri0902
21a4dee575 feat(annotations): query deployment markers across traces and logs
Two-pass resource-table query per signal, merged per tuple on the earliest first seen.
2026-11-01 14:06:14 +05:30
nikhilmantri0902
33405eebd6 feat(annotations): compile resource filter per signal
Exports resourcefilter.NewStorage so the module compiles the filter inline, same as inframonitoring.
2026-11-01 13:13:08 +05:30
nikhilmantri0902
fdf21fda60 chore(annotations): generate openapi spec and frontend client 2026-10-31 20:41:12 +05:30
nikhilmantri0902
daa51df45b feat(annotations): scaffold annotations module and list endpoint
POST /api/v1/annotations returns an empty list until the markers query lands.
2026-10-31 20:41:11 +05:30
34 changed files with 2685 additions and 21 deletions

View File

@@ -1689,6 +1689,108 @@ components:
required:
- config
type: object
AnnotationtypesAnnotation:
discriminator:
mapping:
deployment: '#/components/schemas/AnnotationtypesAnnotationDeployment'
propertyName: kind
oneOf:
- $ref: '#/components/schemas/AnnotationtypesAnnotationDeployment'
type: object
AnnotationtypesAnnotationDeployment:
properties:
kind:
$ref: '#/components/schemas/AnnotationtypesAnnotationKind'
spec:
$ref: '#/components/schemas/AnnotationtypesDeploymentSpec'
required:
- kind
- spec
type: object
AnnotationtypesAnnotationKind:
enum:
- deployment
type: string
AnnotationtypesAnnotations:
properties:
historyAvailableFrom:
format: int64
type: integer
records:
items:
$ref: '#/components/schemas/AnnotationtypesAnnotation'
nullable: true
type: array
total:
type: integer
required:
- records
- total
- historyAvailableFrom
type: object
AnnotationtypesDeploymentSpec:
properties:
environment:
type: string
firstSeen:
format: int64
type: integer
id:
type: string
logId:
type: string
resourceAttributes:
additionalProperties:
type: string
nullable: true
type: object
service:
type: string
signal:
$ref: '#/components/schemas/TelemetrytypesSignal'
spanId:
type: string
traceId:
type: string
version:
type: string
required:
- id
- signal
- service
- environment
- version
- firstSeen
- traceId
- spanId
- logId
- resourceAttributes
type: object
AnnotationtypesPostableAnnotations:
properties:
end:
format: int64
type: integer
filter:
$ref: '#/components/schemas/Querybuildertypesv5Filter'
kind:
$ref: '#/components/schemas/AnnotationtypesAnnotationKind'
limit:
type: integer
signals:
items:
$ref: '#/components/schemas/TelemetrytypesSignal'
nullable: true
type: array
start:
format: int64
type: integer
required:
- kind
- start
- end
- limit
type: object
AuthtypesAttributeMapping:
properties:
email:
@@ -13760,6 +13862,73 @@ paths:
tags:
- alerts
x-signoz-stability: alpha
/api/v1/annotations:
post:
deprecated: false
description: Returns annotation markers of the requested kind inside the time
range. For kind 'deployment', a marker is the first time a (service, environment,
version) tuple was seen in the requested signals (traces, logs, or both when
omitted). Each record carries the signal that produced it, the first seen
time in epoch milliseconds, the trace and span ids or the log id of the first
row, and the resource attributes of that row minus the identity keys. Supports
an optional resource-attribute filter expression. historyAvailableFrom is
the earliest time the result can be trusted, bounded by the signal retention.
operationId: ListAnnotations
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/AnnotationtypesPostableAnnotations'
responses:
"200":
content:
application/json:
schema:
properties:
data:
$ref: '#/components/schemas/AnnotationtypesAnnotations'
status:
type: string
required:
- status
- data
type: object
description: OK
"400":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Bad Request
"401":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Unauthorized
"403":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Forbidden
"500":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
security:
- api_key:
- traces:read
- logs:read
- tokenizer:
- traces:read
- logs:read
summary: List annotations
tags:
- annotations
x-signoz-stability: alpha
/api/v1/authz/check:
post:
deprecated: false

View File

@@ -0,0 +1,105 @@
/**
* ! Do not edit manually
* * The file has been auto-generated using Orval for SigNoz
* * regenerate with 'pnpm generate:api'
* SigNoz
*/
import { useMutation } from 'react-query';
import type {
MutationFunction,
UseMutationOptions,
UseMutationResult,
} from 'react-query';
import type {
AnnotationtypesPostableAnnotationsDTO,
ListAnnotations200,
RenderErrorResponseDTO,
} from '../sigNoz.schemas';
import { GeneratedAPIInstance } from '../../../generatedAPIInstance';
import type { ErrorType, BodyType } from '../../../generatedAPIInstance';
/**
* Returns annotation markers of the requested kind inside the time range. For kind 'deployment', a marker is the first time a (service, environment, version) tuple was seen in the requested signals (traces, logs, or both when omitted). Each record carries the signal that produced it, the first seen time in epoch milliseconds, the trace and span ids or the log id of the first row, and the resource attributes of that row minus the identity keys. Supports an optional resource-attribute filter expression. historyAvailableFrom is the earliest time the result can be trusted, bounded by the signal retention.
* @summary List annotations
*/
export const listAnnotations = (
annotationtypesPostableAnnotationsDTO?: BodyType<AnnotationtypesPostableAnnotationsDTO>,
signal?: AbortSignal,
) => {
return GeneratedAPIInstance<ListAnnotations200>({
url: `/api/v1/annotations`,
method: 'POST',
headers: { 'Content-Type': 'application/json' },
data: annotationtypesPostableAnnotationsDTO,
signal,
});
};
export const getListAnnotationsMutationOptions = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof listAnnotations>>,
TError,
{ data?: BodyType<AnnotationtypesPostableAnnotationsDTO> },
TContext
>;
}): UseMutationOptions<
Awaited<ReturnType<typeof listAnnotations>>,
TError,
{ data?: BodyType<AnnotationtypesPostableAnnotationsDTO> },
TContext
> => {
const mutationKey = ['listAnnotations'];
const { mutation: mutationOptions } = options
? options.mutation &&
'mutationKey' in options.mutation &&
options.mutation.mutationKey
? options
: { ...options, mutation: { ...options.mutation, mutationKey } }
: { mutation: { mutationKey } };
const mutationFn: MutationFunction<
Awaited<ReturnType<typeof listAnnotations>>,
{ data?: BodyType<AnnotationtypesPostableAnnotationsDTO> }
> = (props) => {
const { data } = props ?? {};
return listAnnotations(data);
};
return { mutationFn, ...mutationOptions };
};
export type ListAnnotationsMutationResult = NonNullable<
Awaited<ReturnType<typeof listAnnotations>>
>;
export type ListAnnotationsMutationBody =
| BodyType<AnnotationtypesPostableAnnotationsDTO>
| undefined;
export type ListAnnotationsMutationError = ErrorType<RenderErrorResponseDTO>;
/**
* @summary List annotations
*/
export const useListAnnotations = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof listAnnotations>>,
TError,
{ data?: BodyType<AnnotationtypesPostableAnnotationsDTO> },
TContext
>;
}): UseMutationResult<
Awaited<ReturnType<typeof listAnnotations>>,
TError,
{ data?: BodyType<AnnotationtypesPostableAnnotationsDTO> },
TContext
> => {
return useMutation(getListAnnotationsMutationOptions(options));
};

View File

@@ -3217,6 +3217,127 @@ export interface AlertmanagertypesUpdatableNotificationChannelDTO {
config: AlertmanagertypesChannelConfigDTO;
}
export enum AnnotationtypesAnnotationDeploymentDTOKind {
deployment = 'deployment',
}
export enum TelemetrytypesSignalDTO {
traces = 'traces',
logs = 'logs',
metrics = 'metrics',
'' = '',
}
export type AnnotationtypesDeploymentSpecDTOResourceAttributesAnyOf = {
[key: string]: string;
};
/**
* @nullable
*/
export type AnnotationtypesDeploymentSpecDTOResourceAttributes = AnnotationtypesDeploymentSpecDTOResourceAttributesAnyOf | null;
export interface AnnotationtypesDeploymentSpecDTO {
/**
* @type string
*/
environment: string;
/**
* @type integer
* @format int64
*/
firstSeen: number;
/**
* @type string
*/
id: string;
/**
* @type string
*/
logId: string;
/**
* @type object,null
*/
resourceAttributes: AnnotationtypesDeploymentSpecDTOResourceAttributes;
/**
* @type string
*/
service: string;
signal: TelemetrytypesSignalDTO;
/**
* @type string
*/
spanId: string;
/**
* @type string
*/
traceId: string;
/**
* @type string
*/
version: string;
}
export interface AnnotationtypesAnnotationDeploymentDTO {
/**
* @type string
* @enum deployment
*/
kind: AnnotationtypesAnnotationDeploymentDTOKind;
spec: AnnotationtypesDeploymentSpecDTO;
}
export type AnnotationtypesAnnotationDTO =
AnnotationtypesAnnotationDeploymentDTO;
export enum AnnotationtypesAnnotationKindDTO {
deployment = 'deployment',
}
export interface AnnotationtypesAnnotationsDTO {
/**
* @type integer
* @format int64
*/
historyAvailableFrom: number;
/**
* @type array,null
*/
records: AnnotationtypesAnnotationDTO[] | null;
/**
* @type integer
*/
total: number;
}
export interface Querybuildertypesv5FilterDTO {
/**
* @type string
* @description A filter in the query filter syntax: comparisons such as =, !=, <, >, IN, LIKE, CONTAINS and EXISTS on field names, combined with AND, OR and NOT, with string values in single quotes, for example key = 'value' AND other_key > 10. Which fields exist depends on the signal or endpoint the filter is used with.
*/
expression?: string;
}
export interface AnnotationtypesPostableAnnotationsDTO {
/**
* @type integer
* @format int64
*/
end: number;
filter?: Querybuildertypesv5FilterDTO;
kind: AnnotationtypesAnnotationKindDTO;
/**
* @type integer
*/
limit: number;
/**
* @type array,null
*/
signals?: TelemetrytypesSignalDTO[] | null;
/**
* @type integer
* @format int64
*/
start: number;
}
export interface AuthtypesAttributeMappingDTO {
/**
* @type string
@@ -5012,14 +5133,6 @@ export type Querybuildertypesv5BucketOptionsDTO =
| Querybuildertypesv5BucketOptionsLinearDTO
| Querybuildertypesv5BucketOptionsLogDTO;
export interface Querybuildertypesv5FilterDTO {
/**
* @type string
* @description A filter in the query filter syntax: comparisons such as =, !=, <, >, IN, LIKE, CONTAINS and EXISTS on field names, combined with AND, OR and NOT, with string values in single quotes, for example key = 'value' AND other_key > 10. Which fields exist depends on the signal or endpoint the filter is used with.
*/
expression?: string;
}
/**
* The argument value as a number, string or boolean, for example the threshold for cutOffMin or the duration for timeShift.
*/
@@ -5088,12 +5201,6 @@ export enum TelemetrytypesFieldDataTypeDTO {
number = 'number',
'' = '',
}
export enum TelemetrytypesSignalDTO {
traces = 'traces',
logs = 'logs',
metrics = 'metrics',
'' = '',
}
export interface Querybuildertypesv5GroupByKeyDTO {
/**
* @type string
@@ -14803,6 +14910,14 @@ export type GetAlerts200 = {
status: string;
};
export type ListAnnotations200 = {
data: AnnotationtypesAnnotationsDTO;
/**
* @type string
*/
status: string;
};
export type AuthzCheck200 = {
/**
* @type array

View File

@@ -0,0 +1,51 @@
package signozapiserver
import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/types/annotationtypes"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/types/coretypes"
"github.com/gorilla/mux"
)
func (provider *provider) addAnnotationsRoutes(router *mux.Router) error {
if err := router.Handle("/api/v1/annotations", handler.New(
provider.authzMiddleware.CheckResources(provider.annotationsHandler.ListAnnotations, authtypes.SigNozAdminRoleName, authtypes.SigNozEditorRoleName, authtypes.SigNozViewerRoleName),
handler.OpenAPIDef{
ID: "ListAnnotations",
Tags: []string{"annotations"},
Summary: "List annotations",
Description: "Returns annotation markers of the requested kind inside the time range. For kind 'deployment', a marker is the first time a (service, environment, version) tuple was seen in the requested signals (traces, logs, or both when omitted). Each record carries the signal that produced it, the first seen time in epoch milliseconds, the trace and span ids or the log id of the first row, and the resource attributes of that row minus the identity keys. Supports an optional resource-attribute filter expression. historyAvailableFrom is the earliest time the result can be trusted, bounded by the signal retention.",
Request: new(annotationtypes.PostableAnnotations),
RequestContentType: "application/json",
Response: new(annotationtypes.Annotations),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusUnauthorized},
Deprecated: false,
SecuritySchemes: newScopedSecuritySchemes([]string{
coretypes.ResourceTelemetryResourceTraces.Scope(coretypes.VerbRead),
coretypes.ResourceTelemetryResourceLogs.Scope(coretypes.VerbRead),
}),
}, handler.WithResourceDefs(
handler.TelemetryResourceDef{
Verb: coretypes.VerbRead,
Category: coretypes.ActionCategoryDataAccess,
Selector: querybuilder.TelemetrySelector,
Resources: querybuilder.BuilderQueryWithFilterExpressionResources(coretypes.ResourceTelemetryResourceTraces, "filter.expression"),
},
handler.TelemetryResourceDef{
Verb: coretypes.VerbRead,
Category: coretypes.ActionCategoryDataAccess,
Selector: querybuilder.TelemetrySelector,
Resources: querybuilder.BuilderQueryWithFilterExpressionResources(coretypes.ResourceTelemetryResourceLogs, "filter.expression"),
},
))).Methods(http.MethodPost).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -18,6 +18,7 @@ import (
"github.com/SigNoz/signoz/pkg/identn"
"github.com/SigNoz/signoz/pkg/licensing"
"github.com/SigNoz/signoz/pkg/modules/aiobservability"
"github.com/SigNoz/signoz/pkg/modules/annotations"
"github.com/SigNoz/signoz/pkg/modules/authdomain"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
"github.com/SigNoz/signoz/pkg/modules/dashboard"
@@ -73,6 +74,7 @@ type provider struct {
metricsExplorerHandler metricsexplorer.Handler
metricReductionRuleHandler metricreductionrule.Handler
infraMonitoringHandler inframonitoring.Handler
annotationsHandler annotations.Handler
gatewayHandler gateway.Handler
gatewayService gateway.Gateway
fieldsHandler fields.Handler
@@ -120,6 +122,7 @@ func NewFactory(
metricsExplorerHandler metricsexplorer.Handler,
metricReductionRuleHandler metricreductionrule.Handler,
infraMonitoringHandler inframonitoring.Handler,
annotationsHandler annotations.Handler,
gatewayHandler gateway.Handler,
gatewayService gateway.Gateway,
fieldsHandler fields.Handler,
@@ -175,6 +178,7 @@ func NewFactory(
metricsExplorerHandler,
metricReductionRuleHandler,
infraMonitoringHandler,
annotationsHandler,
gatewayHandler,
gatewayService,
fieldsHandler,
@@ -232,6 +236,7 @@ func newProvider(
metricsExplorerHandler metricsexplorer.Handler,
metricReductionRuleHandler metricreductionrule.Handler,
infraMonitoringHandler inframonitoring.Handler,
annotationsHandler annotations.Handler,
gatewayHandler gateway.Handler,
gatewayService gateway.Gateway,
fieldsHandler fields.Handler,
@@ -289,6 +294,7 @@ func newProvider(
metricsExplorerHandler: metricsExplorerHandler,
metricReductionRuleHandler: metricReductionRuleHandler,
infraMonitoringHandler: infraMonitoringHandler,
annotationsHandler: annotationsHandler,
gatewayHandler: gatewayHandler,
gatewayService: gatewayService,
fieldsHandler: fieldsHandler,
@@ -436,6 +442,10 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
return err
}
if err := provider.addAnnotationsRoutes(router); err != nil {
return err
}
if err := provider.addGatewayRoutes(router); err != nil {
return err
}

View File

@@ -0,0 +1,19 @@
package annotations
import (
"context"
"net/http"
"github.com/SigNoz/signoz/pkg/statsreporter"
"github.com/SigNoz/signoz/pkg/types/annotationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
)
type Handler interface {
ListAnnotations(http.ResponseWriter, *http.Request)
}
type Module interface {
statsreporter.StatsCollector
ListAnnotations(ctx context.Context, orgID valuer.UUID, req *annotationtypes.PostableAnnotations) (*annotationtypes.Annotations, error)
}

View File

@@ -0,0 +1,27 @@
package implannotations
import (
"context"
"fmt"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/valuer"
)
func (m *module) Collect(ctx context.Context, _ valuer.UUID) (map[string]any, error) {
stats := make(map[string]any)
query := fmt.Sprintf(
"SELECT (SELECT count() FROM (SELECT 1 FROM %s.%s WHERE labels LIKE '%%\"service.version\"%%' LIMIT 1)), (SELECT count() FROM (SELECT 1 FROM %s.%s WHERE labels LIKE '%%\"service.version\"%%' LIMIT 1))",
tracesTables.dbName, tracesTables.resourceTable, logsTables.dbName, logsTables.resourceTable,
)
var tracesCount, logsCount uint64
if err := m.telemetryStore.ClickhouseDB().QueryRow(ctx, query).Scan(&tracesCount, &logsCount); err == nil {
stats["telemetry.traces.service_version.exists"] = tracesCount > 0
stats["telemetry.logs.service_version.exists"] = logsCount > 0
} else {
m.logger.DebugContext(ctx, "failed to collect service.version existence stats", errors.Attr(err))
}
return stats, nil
}

View File

@@ -0,0 +1,294 @@
package implannotations
import (
"context"
"fmt"
"strings"
"time"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/semconv"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/types/annotationtypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
)
const (
resourceBucketSeconds = 1800
legacyEnvironmentKey = "deployment.environment"
)
type signalTables struct {
signal telemetrytypes.Signal
dbName string
resourceTable string
dataTable string
rawAlias string
}
var (
tracesTables = signalTables{
signal: telemetrytypes.SignalTraces,
dbName: tracestelemetryschema.DBName,
resourceTable: tracestelemetryschema.TracesResourceV3TableName,
dataTable: tracestelemetryschema.SpanIndexV3TableName,
rawAlias: "raw_traces",
}
logsTables = signalTables{
signal: telemetrytypes.SignalLogs,
dbName: logstelemetryschema.DBName,
resourceTable: logstelemetryschema.LogsResourceV2TableName,
dataTable: logstelemetryschema.LogsV2TableName,
rawAlias: "raw_logs",
}
)
// queryTracesDeployments returns the marker rows and historyAvailableFrom in ms, 0 when there is no trace history.
func (m *module) queryTracesDeployments(ctx context.Context, orgID valuer.UUID, req *annotationtypes.PostableAnnotations) ([]annotationtypes.DeploymentRow, int64, error) {
ctx = m.withAnnotationsContext(ctx, telemetrytypes.SignalTraces, "ListAnnotations")
filterClause, err := m.buildFilterClause(ctx, orgID, telemetrytypes.SignalTraces, req.Filter, req.Start, req.End)
if err != nil {
return nil, 0, err
}
rBucket, err := m.readHistoryEdge(ctx, tracesTables)
if err != nil {
return nil, 0, err
}
if rBucket == 0 {
return []annotationtypes.DeploymentRow{}, 0, nil
}
query, args := buildTracesDeploymentsQuery(filterClause, resolveEnvironmentKeys(ctx, orgID, m.fl, telemetrytypes.SignalTraces), req.Start, req.End, rBucket)
rows, err := m.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
if err != nil {
return nil, 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to query traces deployments")
}
defer rows.Close()
result := []annotationtypes.DeploymentRow{}
for rows.Next() {
var (
row = annotationtypes.DeploymentRow{Signal: telemetrytypes.SignalTraces}
firstSeen time.Time
firstIDs []string
)
if err := rows.Scan(&row.Service, &row.Environment, &row.Version, &firstSeen, &firstIDs, &row.Labels); err != nil {
return nil, 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to scan traces deployment row")
}
row.FirstSeen = firstSeen.UnixMilli()
if len(firstIDs) == 2 {
row.TraceID = firstIDs[0]
row.SpanID = firstIDs[1]
}
result = append(result, row)
}
if err := rows.Err(); err != nil {
return nil, 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to iterate traces deployment rows")
}
return result, historyAvailableFromMs(rBucket), nil
}
// queryLogsDeployments returns the marker rows and historyAvailableFrom in ms, 0 when there is no log history.
func (m *module) queryLogsDeployments(ctx context.Context, orgID valuer.UUID, req *annotationtypes.PostableAnnotations) ([]annotationtypes.DeploymentRow, int64, error) {
ctx = m.withAnnotationsContext(ctx, telemetrytypes.SignalLogs, "ListAnnotations")
filterClause, err := m.buildFilterClause(ctx, orgID, telemetrytypes.SignalLogs, req.Filter, req.Start, req.End)
if err != nil {
return nil, 0, err
}
rBucket, err := m.readHistoryEdge(ctx, logsTables)
if err != nil {
return nil, 0, err
}
if rBucket == 0 {
return []annotationtypes.DeploymentRow{}, 0, nil
}
query, args := buildLogsDeploymentsQuery(filterClause, resolveEnvironmentKeys(ctx, orgID, m.fl, telemetrytypes.SignalLogs), req.Start, req.End, rBucket)
rows, err := m.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
if err != nil {
return nil, 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to query logs deployments")
}
defer rows.Close()
result := []annotationtypes.DeploymentRow{}
for rows.Next() {
var (
row = annotationtypes.DeploymentRow{Signal: telemetrytypes.SignalLogs}
firstSeenNs uint64
)
if err := rows.Scan(&row.Service, &row.Environment, &row.Version, &firstSeenNs, &row.LogID, &row.Labels); err != nil {
return nil, 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to scan logs deployment row")
}
row.FirstSeen = int64(firstSeenNs / 1000000)
result = append(result, row)
}
if err := rows.Err(); err != nil {
return nil, 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to iterate logs deployment rows")
}
return result, historyAvailableFromMs(rBucket), nil
}
// readHistoryEdge returns the oldest bucket the resource table holds, 0 when it is empty.
func (m *module) readHistoryEdge(ctx context.Context, tables signalTables) (int64, error) {
var rBucket int64
if err := m.telemetryStore.ClickhouseDB().QueryRow(ctx, buildHistoryEdgeQuery(tables)).Scan(&rBucket); err != nil {
return 0, errors.WrapInternalf(err, errors.CodeInternal, "failed to read %s retention edge", tables.signal.StringValue())
}
return rBucket, nil
}
func buildTracesDeploymentsQuery(filterClause *sqlbuilder.WhereClause, envKeys []string, startMs, endMs, rBucket int64) (string, []any) {
bounds := []string{
"fromUnixTimestamp64Milli(toInt64(?)) AS t1_dt",
"fromUnixTimestamp64Milli(toInt64(?)) AS t2_dt",
}
ctes, cteArgs := buildResourcePasses(tracesTables, filterClause, envKeys, startMs, endMs, rBucket, bounds, []any{startMs, endMs})
final := buildFinalSelect(
tracesTables,
"argMin((raw_traces.trace_id, raw_traces.span_id), raw_traces.timestamp) AS first_ids",
"first_seen >= t1_dt AND first_seen <= t2_dt",
)
return querybuilder.CombineCTEs(ctes) + final, querybuilder.PrependArgs(cteArgs, nil)
}
func buildLogsDeploymentsQuery(filterClause *sqlbuilder.WhereClause, envKeys []string, startMs, endMs, rBucket int64) (string, []any) {
bounds := []string{
"toUInt64(?) * 1000000 AS t1_ns",
"toUInt64(?) * 1000000 AS t2_ns",
}
ctes, cteArgs := buildResourcePasses(logsTables, filterClause, envKeys, startMs, endMs, rBucket, bounds, []any{startMs, endMs})
final := buildFinalSelect(
logsTables,
"argMin(raw_logs.id, raw_logs.timestamp) AS first_id",
"first_seen >= t1_ns AND first_seen <= t2_ns",
)
return querybuilder.CombineCTEs(ctes) + final, querybuilder.PrependArgs(cteArgs, nil)
}
// buildResourcePasses returns the constants, the signal's time bounds and the two resource-table CTEs shared by both signals, with their args in order.
func buildResourcePasses(tables signalTables, filterClause *sqlbuilder.WhereClause, envKeys []string, startMs, endMs, rBucket int64, bounds []string, boundArgs []any) ([]string, [][]any) {
constants := []string{
"toInt64(intDiv(?, 1000)) AS t1",
"toInt64(intDiv(?, 1000)) AS t2",
fmt.Sprintf("toInt64(intDiv(t1, %d) * %d) AS bt1", resourceBucketSeconds, resourceBucketSeconds),
"toInt64(?) AS r_bucket",
}
constants = append(constants, bounds...)
constantArgs := append([]any{startMs, endMs, rBucket}, boundArgs...)
resourceTable := fmt.Sprintf("%s.%s", tables.dbName, tables.resourceTable)
firstBucketInner := sqlbuilder.NewSelectBuilder()
firstBucketInner.Select(
"simpleJSONExtractString(labels, 'service.name') AS service",
buildEnvironmentExpr(envKeys)+" AS env",
"simpleJSONExtractString(labels, 'service.version') AS version",
"seen_at_ts_bucket_start AS bucket",
)
firstBucketInner.From(resourceTable)
firstBucketInner.Where("version != ''", "service != ''", "seen_at_ts_bucket_start <= t2")
firstBucketInner.AddWhereClause(sqlbuilder.CopyWhereClause(filterClause))
firstBucketInnerSQL, firstBucketArgs := firstBucketInner.BuildWithFlavor(sqlbuilder.ClickHouse)
firstBucketSQL := fmt.Sprintf(
"SELECT service, env, version, min(bucket) AS min_bucket FROM (%s) GROUP BY service, env, version HAVING min_bucket >= bt1 AND min_bucket > r_bucket",
firstBucketInnerSQL,
)
cand := sqlbuilder.NewSelectBuilder()
cand.Select(
"simpleJSONExtractString(labels, 'service.name') AS service",
buildEnvironmentExpr(envKeys)+" AS env",
"simpleJSONExtractString(labels, 'service.version') AS version",
"seen_at_ts_bucket_start AS min_bucket",
"fingerprint AS fp",
"labels",
)
cand.From(resourceTable)
cand.Where(
"version != ''",
"service != ''",
"(service, env, version, seen_at_ts_bucket_start) GLOBAL IN (SELECT service, env, version, min_bucket FROM first_bucket)",
)
cand.AddWhereClause(sqlbuilder.CopyWhereClause(filterClause))
candSQL, candArgs := cand.BuildWithFlavor(sqlbuilder.ClickHouse)
ctes := make([]string, 0, len(constants)+2)
ctes = append(ctes, constants...)
ctes = append(ctes, fmt.Sprintf("first_bucket AS (%s)", firstBucketSQL), fmt.Sprintf("cand AS (%s)", candSQL))
return ctes, [][]any{constantArgs, firstBucketArgs, candArgs}
}
// buildFinalSelect joins the raw table back to cand; idSelect and having carry the signal-specific columns and bounds.
// resolveEnvironmentKeys returns the environment spellings to read, the semconv family when the families flag is on, else the legacy key alone.
func resolveEnvironmentKeys(ctx context.Context, orgID valuer.UUID, fl flagger.Flagger, signal telemetrytypes.Signal) []string {
if !querybuilder.SemconvFamiliesEnabled(ctx, orgID, fl) {
return []string{legacyEnvironmentKey}
}
return semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
Name: legacyEnvironmentKey,
Signal: signal,
FieldContext: telemetrytypes.FieldContextResource,
})
}
// buildEnvironmentExpr mirrors the string family read the filter path compiles: first non-empty spelling wins, ” when none is set.
func buildEnvironmentExpr(keys []string) string {
if len(keys) == 1 {
return fmt.Sprintf("simpleJSONExtractString(labels, '%s')", keys[0])
}
reads := make([]string, 0, len(keys))
for _, key := range keys {
reads = append(reads, fmt.Sprintf("NULLIF(simpleJSONExtractString(labels, '%s'), '')", key))
}
return fmt.Sprintf("COALESCE(%s, '')", strings.Join(reads, ", "))
}
func buildFinalSelect(tables signalTables, idSelect, having string) string {
raw := tables.rawAlias
return fmt.Sprintf(
"SELECT cand.service, cand.env, cand.version, min(%s.timestamp) AS first_seen, %s, argMin(cand.labels, %s.timestamp) AS labels "+
"FROM %s.%s AS %s "+
"GLOBAL INNER JOIN cand ON %s.resource_fingerprint = cand.fp AND toInt64(%s.ts_bucket_start) = cand.min_bucket "+
"WHERE (%s.ts_bucket_start, %s.resource_fingerprint) GLOBAL IN (SELECT toUInt64(min_bucket), fp FROM cand) "+
"GROUP BY cand.service, cand.env, cand.version "+
"HAVING %s "+
"ORDER BY first_seen",
raw, idSelect, raw,
tables.dbName, tables.dataTable, raw,
raw, raw,
raw, raw,
having,
)
}
func buildHistoryEdgeQuery(tables signalTables) string {
return fmt.Sprintf("SELECT min(seen_at_ts_bucket_start) FROM %s.%s", tables.dbName, tables.resourceTable)
}
func historyAvailableFromMs(rBucket int64) int64 {
return (rBucket + resourceBucketSeconds) * 1000
}
// resolveSignals returns the requested signals, both when none were given.
func resolveSignals(req *annotationtypes.PostableAnnotations) []telemetrytypes.Signal {
if len(req.Signals) == 0 {
return []telemetrytypes.Signal{telemetrytypes.SignalTraces, telemetrytypes.SignalLogs}
}
return req.Signals
}

View File

@@ -0,0 +1,200 @@
package implannotations
import (
"context"
"strings"
"testing"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/types/annotationtypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func serviceFilterClause(value string) *sqlbuilder.WhereClause {
clause := sqlbuilder.NewWhereClause()
cond := sqlbuilder.NewCond()
clause.AddWhereExpr(cond.Args, cond.Equal("simpleJSONExtractString(labels, 'service.name')", value))
return clause
}
var (
legacyEnvKeys = []string{"deployment.environment"}
familyEnvKeys = []string{"deployment.environment.name", "deployment.environment"}
familyEnvExpr = "COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '')"
)
// withFamilyEnv swaps the legacy environment read for the coalesced family read in both resource passes of a golden.
func withFamilyEnv(golden string) string {
return strings.Replace(golden, "simpleJSONExtractString(labels, 'deployment.environment') AS env", familyEnvExpr+" AS env", 2)
}
// withServiceFilter inserts the service predicate into both resource passes of an unfiltered golden.
func withServiceFilter(golden string) string {
golden = strings.Replace(golden, "seen_at_ts_bucket_start <= t2)", "seen_at_ts_bucket_start <= t2 AND simpleJSONExtractString(labels, 'service.name') = ?)", 1)
return strings.Replace(golden, "min_bucket FROM first_bucket))", "min_bucket FROM first_bucket) AND simpleJSONExtractString(labels, 'service.name') = ?)", 1)
}
func TestBuildTracesDeploymentsQuery(t *testing.T) {
noFilter := "WITH toInt64(intDiv(?, 1000)) AS t1, toInt64(intDiv(?, 1000)) AS t2, toInt64(intDiv(t1, 1800) * 1800) AS bt1, toInt64(?) AS r_bucket, fromUnixTimestamp64Milli(toInt64(?)) AS t1_dt, fromUnixTimestamp64Milli(toInt64(?)) AS t2_dt, " +
"first_bucket AS (SELECT service, env, version, min(bucket) AS min_bucket FROM (SELECT simpleJSONExtractString(labels, 'service.name') AS service, simpleJSONExtractString(labels, 'deployment.environment') AS env, simpleJSONExtractString(labels, 'service.version') AS version, seen_at_ts_bucket_start AS bucket FROM signoz_traces.distributed_traces_v3_resource WHERE version != '' AND service != '' AND seen_at_ts_bucket_start <= t2) GROUP BY service, env, version HAVING min_bucket >= bt1 AND min_bucket > r_bucket), " +
"cand AS (SELECT simpleJSONExtractString(labels, 'service.name') AS service, simpleJSONExtractString(labels, 'deployment.environment') AS env, simpleJSONExtractString(labels, 'service.version') AS version, seen_at_ts_bucket_start AS min_bucket, fingerprint AS fp, labels FROM signoz_traces.distributed_traces_v3_resource WHERE version != '' AND service != '' AND (service, env, version, seen_at_ts_bucket_start) GLOBAL IN (SELECT service, env, version, min_bucket FROM first_bucket)) " +
"SELECT cand.service, cand.env, cand.version, min(raw_traces.timestamp) AS first_seen, argMin((raw_traces.trace_id, raw_traces.span_id), raw_traces.timestamp) AS first_ids, argMin(cand.labels, raw_traces.timestamp) AS labels FROM signoz_traces.distributed_signoz_index_v3 AS raw_traces GLOBAL INNER JOIN cand ON raw_traces.resource_fingerprint = cand.fp AND toInt64(raw_traces.ts_bucket_start) = cand.min_bucket WHERE (raw_traces.ts_bucket_start, raw_traces.resource_fingerprint) GLOBAL IN (SELECT toUInt64(min_bucket), fp FROM cand) GROUP BY cand.service, cand.env, cand.version HAVING first_seen >= t1_dt AND first_seen <= t2_dt ORDER BY first_seen"
testCases := []struct {
name string
filter *sqlbuilder.WhereClause
envKeys []string
wantSQL string
wantArgs []any
}{
{
name: "NoFilter_Golden",
filter: sqlbuilder.NewWhereClause(),
envKeys: legacyEnvKeys,
wantSQL: noFilter,
wantArgs: []any{int64(1700000000000), int64(1700003600000), int64(1699990000), int64(1700000000000), int64(1700003600000)},
},
{
name: "Filtered_ArgsInBothPasses",
filter: serviceFilterClause("cart"),
envKeys: legacyEnvKeys,
wantSQL: withServiceFilter(noFilter),
wantArgs: []any{int64(1700000000000), int64(1700003600000), int64(1699990000), int64(1700000000000), int64(1700003600000), "cart", "cart"},
},
{
name: "FamilyKeys_CoalescedEnvInBothPasses",
filter: sqlbuilder.NewWhereClause(),
envKeys: familyEnvKeys,
wantSQL: withFamilyEnv(noFilter),
wantArgs: []any{int64(1700000000000), int64(1700003600000), int64(1699990000), int64(1700000000000), int64(1700003600000)},
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
sql, args := buildTracesDeploymentsQuery(testCase.filter, testCase.envKeys, 1700000000000, 1700003600000, 1699990000)
assert.Equal(t, testCase.wantSQL, sql)
assert.Equal(t, testCase.wantArgs, args)
assert.Equal(t, strings.Count(sql, "?"), len(args))
})
}
}
func TestBuildLogsDeploymentsQuery(t *testing.T) {
noFilter := "WITH toInt64(intDiv(?, 1000)) AS t1, toInt64(intDiv(?, 1000)) AS t2, toInt64(intDiv(t1, 1800) * 1800) AS bt1, toInt64(?) AS r_bucket, toUInt64(?) * 1000000 AS t1_ns, toUInt64(?) * 1000000 AS t2_ns, " +
"first_bucket AS (SELECT service, env, version, min(bucket) AS min_bucket FROM (SELECT simpleJSONExtractString(labels, 'service.name') AS service, simpleJSONExtractString(labels, 'deployment.environment') AS env, simpleJSONExtractString(labels, 'service.version') AS version, seen_at_ts_bucket_start AS bucket FROM signoz_logs.distributed_logs_v2_resource WHERE version != '' AND service != '' AND seen_at_ts_bucket_start <= t2) GROUP BY service, env, version HAVING min_bucket >= bt1 AND min_bucket > r_bucket), " +
"cand AS (SELECT simpleJSONExtractString(labels, 'service.name') AS service, simpleJSONExtractString(labels, 'deployment.environment') AS env, simpleJSONExtractString(labels, 'service.version') AS version, seen_at_ts_bucket_start AS min_bucket, fingerprint AS fp, labels FROM signoz_logs.distributed_logs_v2_resource WHERE version != '' AND service != '' AND (service, env, version, seen_at_ts_bucket_start) GLOBAL IN (SELECT service, env, version, min_bucket FROM first_bucket)) " +
"SELECT cand.service, cand.env, cand.version, min(raw_logs.timestamp) AS first_seen, argMin(raw_logs.id, raw_logs.timestamp) AS first_id, argMin(cand.labels, raw_logs.timestamp) AS labels FROM signoz_logs.distributed_logs_v2 AS raw_logs GLOBAL INNER JOIN cand ON raw_logs.resource_fingerprint = cand.fp AND toInt64(raw_logs.ts_bucket_start) = cand.min_bucket WHERE (raw_logs.ts_bucket_start, raw_logs.resource_fingerprint) GLOBAL IN (SELECT toUInt64(min_bucket), fp FROM cand) GROUP BY cand.service, cand.env, cand.version HAVING first_seen >= t1_ns AND first_seen <= t2_ns ORDER BY first_seen"
testCases := []struct {
name string
filter *sqlbuilder.WhereClause
envKeys []string
wantSQL string
wantArgs []any
}{
{
name: "NoFilter_Golden",
filter: sqlbuilder.NewWhereClause(),
envKeys: legacyEnvKeys,
wantSQL: noFilter,
wantArgs: []any{int64(1700000000000), int64(1700003600000), int64(1699990000), int64(1700000000000), int64(1700003600000)},
},
{
name: "Filtered_ArgsInBothPasses",
filter: serviceFilterClause("cart"),
envKeys: legacyEnvKeys,
wantSQL: withServiceFilter(noFilter),
wantArgs: []any{int64(1700000000000), int64(1700003600000), int64(1699990000), int64(1700000000000), int64(1700003600000), "cart", "cart"},
},
{
name: "FamilyKeys_CoalescedEnvInBothPasses",
filter: sqlbuilder.NewWhereClause(),
envKeys: familyEnvKeys,
wantSQL: withFamilyEnv(noFilter),
wantArgs: []any{int64(1700000000000), int64(1700003600000), int64(1699990000), int64(1700000000000), int64(1700003600000)},
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
sql, args := buildLogsDeploymentsQuery(testCase.filter, testCase.envKeys, 1700000000000, 1700003600000, 1699990000)
assert.Equal(t, testCase.wantSQL, sql)
assert.Equal(t, testCase.wantArgs, args)
assert.Equal(t, strings.Count(sql, "?"), len(args))
})
}
}
func TestBuildEnvironmentExpr(t *testing.T) {
testCases := []struct {
name string
keys []string
want string
}{
{name: "SingleKey_PlainRead", keys: legacyEnvKeys, want: "simpleJSONExtractString(labels, 'deployment.environment')"},
{name: "FamilyKeys_CoalesceCurrentFirst", keys: familyEnvKeys, want: familyEnvExpr},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
assert.Equal(t, testCase.want, buildEnvironmentExpr(testCase.keys))
})
}
}
func TestResolveEnvironmentKeys(t *testing.T) {
testCases := []struct {
name string
familyOn bool
signal telemetrytypes.Signal
want []string
}{
{name: "FlagOff_LegacyOnly", familyOn: false, signal: telemetrytypes.SignalTraces, want: legacyEnvKeys},
{name: "FlagOn_Traces_CurrentThenLegacy", familyOn: true, signal: telemetrytypes.SignalTraces, want: familyEnvKeys},
{name: "FlagOn_Logs_CurrentThenLegacy", familyOn: true, signal: telemetrytypes.SignalLogs, want: familyEnvKeys},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
fl := flaggertest.New(t)
if testCase.familyOn {
fl = flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureResolveSemconvFamilies.String(): true})
}
assert.Equal(t, testCase.want, resolveEnvironmentKeys(context.Background(), valuer.GenerateUUID(), fl, testCase.signal))
})
}
}
func TestBuildHistoryEdgeQuery(t *testing.T) {
assert.Equal(t, "SELECT min(seen_at_ts_bucket_start) FROM signoz_traces.distributed_traces_v3_resource", buildHistoryEdgeQuery(tracesTables))
assert.Equal(t, "SELECT min(seen_at_ts_bucket_start) FROM signoz_logs.distributed_logs_v2_resource", buildHistoryEdgeQuery(logsTables))
}
func TestHistoryAvailableFromMs(t *testing.T) {
assert.Equal(t, int64(1700001800000), historyAvailableFromMs(1700000000))
}
func TestResolveSignals(t *testing.T) {
testCases := []struct {
name string
in []telemetrytypes.Signal
want []telemetrytypes.Signal
}{
{name: "Empty_Both", in: nil, want: []telemetrytypes.Signal{telemetrytypes.SignalTraces, telemetrytypes.SignalLogs}},
{name: "LogsOnly_Logs", in: []telemetrytypes.Signal{telemetrytypes.SignalLogs}, want: []telemetrytypes.Signal{telemetrytypes.SignalLogs}},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
got := resolveSignals(&annotationtypes.PostableAnnotations{Signals: testCase.in})
require.Len(t, got, len(testCase.want))
assert.Equal(t, testCase.want, got)
})
}
}

View File

@@ -0,0 +1,46 @@
package implannotations
import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/SigNoz/signoz/pkg/modules/annotations"
"github.com/SigNoz/signoz/pkg/types/annotationtypes"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/valuer"
)
type handler struct {
module annotations.Module
}
func NewHandler(m annotations.Module) annotations.Handler {
return &handler{
module: m,
}
}
func (h *handler) ListAnnotations(rw http.ResponseWriter, req *http.Request) {
claims, err := authtypes.ClaimsFromContext(req.Context())
if err != nil {
render.Error(rw, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
var parsedReq annotationtypes.PostableAnnotations
if err := binding.JSON.BindBody(req.Body, &parsedReq); err != nil {
render.Error(rw, err)
return
}
result, err := h.module.ListAnnotations(req.Context(), orgID, &parsedReq)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, result)
}

View File

@@ -0,0 +1,55 @@
package implannotations
import (
"context"
"strings"
"github.com/SigNoz/signoz/pkg/querybuilder"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
)
// Returns an empty clause when the expression is empty or holds no resource condition.
func (m *module) buildFilterClause(ctx context.Context, orgID valuer.UUID, signal telemetrytypes.Signal, filter *qbtypes.Filter, startMillis, endMillis int64) (*sqlbuilder.WhereClause, error) {
expression := ""
if filter != nil {
expression = strings.TrimSpace(filter.Expression)
}
if expression == "" {
return sqlbuilder.NewWhereClause(), nil
}
whereClauseSelectors := querybuilder.QueryStringToKeysSelectors(expression)
for idx := range whereClauseSelectors {
whereClauseSelectors[idx].Signal = signal
whereClauseSelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
}
whereClauseSelectors = querybuilder.ExpandKeySelectorsForFamilies(ctx, orgID, m.fl, whereClauseSelectors)
keys, _, err := m.telemetryMetadataStore.GetKeysMulti(ctx, orgID, whereClauseSelectors)
if err != nil {
return nil, err
}
opts := querybuilder.FilterExprVisitorOpts{
Context: ctx,
Query: querybuilder.NewQueryInfo(ctx, orgID, m.fl, signal, nil, querybuilder.ToNanoSecs(uint64(startMillis)), querybuilder.ToNanoSecs(uint64(endMillis))),
Storage: m.storage,
Logger: m.logger,
FieldKeys: keys,
SkipFullTextFilter: true,
}
whereClause, err := querybuilder.PrepareWhereClause(expression, opts)
if err != nil {
return nil, err
}
if whereClause.IsEmpty() {
return sqlbuilder.NewWhereClause(), nil
}
return whereClause.WhereClause, nil
}

View File

@@ -0,0 +1,165 @@
package implannotations
import (
"context"
"strings"
"testing"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func newTestModule(t *testing.T, fl flagger.Flagger, signal telemetrytypes.Signal) *module {
t.Helper()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = map[string][]*telemetrytypes.TelemetryFieldKey{
"service.name": {{
Name: "service.name",
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
"k8s.namespace.name": {{
Name: "k8s.namespace.name",
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
"deployment.environment": {{
Name: "deployment.environment",
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
"deployment.environment.name": {{
Name: "deployment.environment.name",
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
"http.request.method": {{
Name: "http.request.method",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
}},
}
for _, keys := range mockMetadataStore.KeysMap {
for _, key := range keys {
key.Signal = signal
}
}
return &module{
telemetryMetadataStore: mockMetadataStore,
storage: resourcefilter.NewStorage(),
fl: fl,
logger: instrumentationtest.New().ToProviderSettings().Logger,
}
}
func buildSQL(clause *sqlbuilder.WhereClause) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("fingerprint")
sb.From("t")
sb.AddWhereClause(clause)
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return strings.TrimSpace(sql), args
}
func TestBuildFilterClause(t *testing.T) {
testCases := []struct {
name string
signal telemetrytypes.Signal
filter *qbtypes.Filter
familyOn bool
wantSQL string
wantArgs []any
}{
{
name: "NilFilter_EmptyClause",
signal: telemetrytypes.SignalTraces,
filter: nil,
wantSQL: "SELECT fingerprint FROM t",
wantArgs: nil,
},
{
name: "WhitespaceExpression_EmptyClause",
signal: telemetrytypes.SignalTraces,
filter: &qbtypes.Filter{Expression: " "},
wantSQL: "SELECT fingerprint FROM t",
wantArgs: nil,
},
{
name: "ResourceKey_Traces_Compiled",
signal: telemetrytypes.SignalTraces,
filter: &qbtypes.Filter{Expression: "service.name = 'cartservice'"},
wantSQL: "SELECT fingerprint FROM t WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?)",
wantArgs: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%"},
},
{
name: "ResourceKey_Logs_Compiled",
signal: telemetrytypes.SignalLogs,
filter: &qbtypes.Filter{Expression: "k8s.namespace.name = 'shop'"},
wantSQL: "SELECT fingerprint FROM t WHERE (simpleJSONExtractString(labels, 'k8s.namespace.name') = ? AND labels LIKE ? AND labels LIKE ?)",
wantArgs: []any{"shop", "%k8s.namespace.name%", "%k8s.namespace.name\":\"shop%"},
},
{
name: "InList_Compiled",
signal: telemetrytypes.SignalTraces,
filter: &qbtypes.Filter{Expression: "service.name IN ('frontend', 'checkout')"},
wantSQL: "SELECT fingerprint FROM t WHERE ((simpleJSONExtractString(labels, 'service.name') = ? OR simpleJSONExtractString(labels, 'service.name') = ?) AND labels LIKE ? AND (labels LIKE ? OR labels LIKE ?))",
wantArgs: []any{"frontend", "checkout", "%service.name%", "%service.name\":\"frontend%", "%service.name\":\"checkout%"},
},
{
name: "SpanAttributeOnly_EmptyClause",
signal: telemetrytypes.SignalTraces,
filter: &qbtypes.Filter{Expression: "http.request.method = 'GET'"},
wantSQL: "SELECT fingerprint FROM t",
wantArgs: nil,
},
{
name: "SpecialCharacters_BoundAsArgs",
signal: telemetrytypes.SignalTraces,
filter: &qbtypes.Filter{Expression: "service.name = 'ca`rt$0'"},
wantSQL: "SELECT fingerprint FROM t WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?)",
wantArgs: []any{"ca`rt$0", "%service.name%", "%service.name\":\"ca`rt$0%"},
},
{
name: "EnvironmentFamily_FlagOn_Widened",
signal: telemetrytypes.SignalTraces,
filter: &qbtypes.Filter{Expression: "deployment.environment = 'production'"},
familyOn: true,
wantSQL: "SELECT fingerprint FROM t WHERE (COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') = ? AND (labels LIKE ? OR labels LIKE ?) AND (labels LIKE ? OR labels LIKE ?))",
wantArgs: []any{"production", "%deployment.environment.name%", "%deployment.environment%", "%deployment.environment.name\":\"production%", "%deployment.environment\":\"production%"},
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
fl := flaggertest.New(t)
if testCase.familyOn {
fl = flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureResolveSemconvFamilies.String(): true})
}
m := newTestModule(t, fl, testCase.signal)
clause, err := m.buildFilterClause(context.Background(), valuer.UUID{}, testCase.signal, testCase.filter, 1000, 2000)
require.NoError(t, err)
require.NotNil(t, clause)
sql, args := buildSQL(clause)
assert.Equal(t, testCase.wantSQL, sql)
assert.Equal(t, testCase.wantArgs, args)
})
}
}
func TestBuildFilterClause_ParseError(t *testing.T) {
m := newTestModule(t, flaggertest.New(t), telemetrytypes.SignalTraces)
_, err := m.buildFilterClause(context.Background(), valuer.UUID{}, telemetrytypes.SignalTraces, &qbtypes.Filter{Expression: "service.name = "}, 1000, 2000)
assert.Error(t, err)
}

View File

@@ -0,0 +1,102 @@
package implannotations
import (
"context"
"log/slog"
"slices"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/modules/annotations"
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/annotationtypes"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/sync/errgroup"
)
type module struct {
telemetryStore telemetrystore.TelemetryStore
telemetryMetadataStore telemetrytypes.MetadataStore
storage qbtypes.Storage
fl flagger.Flagger
logger *slog.Logger
}
func NewModule(
telemetryStore telemetrystore.TelemetryStore,
telemetryMetadataStore telemetrytypes.MetadataStore,
fl flagger.Flagger,
providerSettings factory.ProviderSettings,
) annotations.Module {
return &module{
telemetryStore: telemetryStore,
telemetryMetadataStore: telemetryMetadataStore,
storage: resourcefilter.NewStorage(),
fl: fl,
logger: providerSettings.Logger,
}
}
func (m *module) ListAnnotations(ctx context.Context, orgID valuer.UUID, req *annotationtypes.PostableAnnotations) (*annotationtypes.Annotations, error) {
if err := req.Validate(); err != nil {
return nil, err
}
var (
tracesSpecs, logsSpecs []annotationtypes.DeploymentSpec
tracesHistoryEdge, logsHistoryEdge int64
)
signals := resolveSignals(req)
errGroup, gCtx := errgroup.WithContext(ctx)
if slices.Contains(signals, telemetrytypes.SignalTraces) {
errGroup.Go(func() error {
rows, edge, err := m.queryTracesDeployments(gCtx, orgID, req)
if err != nil {
return err
}
tracesSpecs, err = annotationtypes.NewDeploymentSpecsFromRows(rows)
if err != nil {
return err
}
tracesHistoryEdge = edge
return nil
})
}
if slices.Contains(signals, telemetrytypes.SignalLogs) {
errGroup.Go(func() error {
rows, edge, err := m.queryLogsDeployments(gCtx, orgID, req)
if err != nil {
return err
}
logsSpecs, err = annotationtypes.NewDeploymentSpecsFromRows(rows)
if err != nil {
return err
}
logsHistoryEdge = edge
return nil
})
}
if err := errGroup.Wait(); err != nil {
return nil, err
}
page, total := annotationtypes.ApplyLimit(annotationtypes.MergeDeployments(tracesSpecs, logsSpecs), req.Limit)
return annotationtypes.NewAnnotationsFromDeployments(page, total, annotationtypes.EarliestHistoryEdge(tracesHistoryEdge, logsHistoryEdge)), nil
}
func (m *module) withAnnotationsContext(ctx context.Context, signal telemetrytypes.Signal, functionName string) context.Context {
comments := map[string]string{
instrumentationtypes.TelemetrySignal: signal.StringValue(),
instrumentationtypes.CodeNamespace: "annotations",
instrumentationtypes.CodeFunctionName: functionName,
}
return ctxtypes.NewContextWithCommentVals(ctx, comments)
}

View File

@@ -14,6 +14,8 @@ import (
"github.com/SigNoz/signoz/pkg/licensing"
"github.com/SigNoz/signoz/pkg/modules/aiobservability"
"github.com/SigNoz/signoz/pkg/modules/aiobservability/implaiobservability"
"github.com/SigNoz/signoz/pkg/modules/annotations"
"github.com/SigNoz/signoz/pkg/modules/annotations/implannotations"
"github.com/SigNoz/signoz/pkg/modules/apdex"
"github.com/SigNoz/signoz/pkg/modules/apdex/implapdex"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
@@ -72,6 +74,7 @@ type Handlers struct {
MetricsExplorer metricsexplorer.Handler
MetricReductionRule metricreductionrule.Handler
InfraMonitoring inframonitoring.Handler
Annotations annotations.Handler
Global global.Handler
FlaggerHandler flagger.Handler
GatewayHandler gateway.Handler
@@ -125,6 +128,7 @@ func NewHandlers(
MetricsExplorer: implmetricsexplorer.NewHandler(modules.MetricsExplorer),
MetricReductionRule: implmetricreductionrule.NewHandler(modules.MetricReductionRule),
InfraMonitoring: implinframonitoring.NewHandler(modules.InfraMonitoring),
Annotations: implannotations.NewHandler(modules.Annotations),
SpanPercentile: implspanpercentile.NewHandler(modules.SpanPercentile),
Global: signozglobal.NewHandler(global),
FlaggerHandler: flagger.NewHandler(flaggerService),

View File

@@ -9,6 +9,8 @@ import (
"github.com/SigNoz/signoz/pkg/emailing"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/modules/annotations"
"github.com/SigNoz/signoz/pkg/modules/annotations/implannotations"
"github.com/SigNoz/signoz/pkg/modules/apdex"
"github.com/SigNoz/signoz/pkg/modules/apdex/implapdex"
"github.com/SigNoz/signoz/pkg/modules/authdomain"
@@ -85,6 +87,7 @@ type Modules struct {
MetricsExplorer metricsexplorer.Module
MetricReductionRule metricreductionrule.Module
InfraMonitoring inframonitoring.Module
Annotations annotations.Module
Promote promote.Module
ServiceAccount serviceaccount.Module
ServiceAccountGetter serviceaccount.Getter
@@ -155,6 +158,7 @@ func NewModules(
MetricsExplorer: implmetricsexplorer.NewModule(telemetryStore, telemetryMetadataStore, cache, ruleStore, dashboard, fl, providerSettings, config.MetricsExplorer),
MetricReductionRule: metricReductionRule,
InfraMonitoring: implinframonitoring.NewModule(telemetryStore, telemetryMetadataStore, querier, fl, providerSettings, config.InfraMonitoring),
Annotations: implannotations.NewModule(telemetryStore, telemetryMetadataStore, fl, providerSettings),
Promote: implpromote.NewModule(telemetryMetadataStore, telemetryStore, providerSettings),
ServiceAccount: serviceAccount,
ServiceAccountGetter: serviceAccountGetter,

View File

@@ -21,6 +21,7 @@ import (
"github.com/SigNoz/signoz/pkg/instrumentation"
"github.com/SigNoz/signoz/pkg/licensing"
"github.com/SigNoz/signoz/pkg/modules/aiobservability"
"github.com/SigNoz/signoz/pkg/modules/annotations"
"github.com/SigNoz/signoz/pkg/modules/authdomain"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
"github.com/SigNoz/signoz/pkg/modules/dashboard"
@@ -82,6 +83,7 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
struct{ metricsexplorer.Handler }{},
struct{ metricreductionrule.Handler }{},
struct{ inframonitoring.Handler }{},
struct{ annotations.Handler }{},
struct{ gateway.Handler }{},
struct{ gateway.Gateway }{},
struct{ fields.Handler }{},

View File

@@ -353,6 +353,7 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
handlers.MetricsExplorer,
handlers.MetricReductionRule,
handlers.InfraMonitoring,
handlers.Annotations,
handlers.GatewayHandler,
gatewayService,
handlers.Fields,

View File

@@ -570,6 +570,7 @@ func New(
cloudIntegrationModule,
modules.LogsPipeline,
modules.InfraMonitoring,
modules.Annotations,
querier,
authz,
}

View File

@@ -200,7 +200,7 @@ func TestConditionBuilder(t *testing.T) {
},
}
storage := newStorage()
storage := NewStorage()
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()

View File

@@ -20,7 +20,7 @@ import (
func TestFamilyEqualWidensIndexHintsToAnyMember(t *testing.T) {
fl := flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureResolveSemconvFamilies.String(): true})
storage := newStorage()
storage := NewStorage()
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"deployment.environment.name": {{
Name: "deployment.environment.name",
@@ -48,7 +48,7 @@ func TestFamilyEqualWidensIndexHintsToAnyMember(t *testing.T) {
func TestFamilyNotEqualDropsNegatedValueHint(t *testing.T) {
fl := flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureResolveSemconvFamilies.String(): true})
storage := newStorage()
storage := NewStorage()
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"deployment.environment.name": {{
Name: "deployment.environment.name",
@@ -76,7 +76,7 @@ func TestFamilyNotEqualDropsNegatedValueHint(t *testing.T) {
func TestFamilyExistsIsAnyMemberPresence(t *testing.T) {
fl := flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureResolveSemconvFamilies.String(): true})
storage := newStorage()
storage := NewStorage()
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"deployment.environment.name": {{
Name: "deployment.environment.name",
@@ -115,7 +115,7 @@ func TestFamilyExistsIsAnyMemberPresence(t *testing.T) {
// including the negated value hint on !=.
func TestSingleMemberShapesUnchanged(t *testing.T) {
fl := flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureResolveSemconvFamilies.String(): true})
storage := newStorage()
storage := NewStorage()
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
"deployment.environment.name": {{
Name: "deployment.environment.name",

View File

@@ -49,7 +49,7 @@ func New[T any](
logger: set.Logger(),
dbName: dbName,
tableName: tableName,
storage: newStorage(),
storage: NewStorage(),
metadataStore: metadataStore,
signal: signal,
source: source,

View File

@@ -22,7 +22,7 @@ type storage struct{}
var _ qbtypes.Storage = (*storage)(nil)
func newStorage() *storage {
func NewStorage() qbtypes.Storage {
return &storage{}
}

View File

@@ -0,0 +1,125 @@
package annotationtypes
import (
"encoding/json"
"slices"
"github.com/SigNoz/signoz/pkg/errors"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/swaggest/jsonschema-go"
)
var ValidSignals = []telemetrytypes.Signal{telemetrytypes.SignalTraces, telemetrytypes.SignalLogs}
type AnnotationKind struct {
valuer.String
}
var (
AnnotationKindDeployment = AnnotationKind{valuer.NewString("deployment")}
)
func (AnnotationKind) Enum() []any {
return []any{
AnnotationKindDeployment,
}
}
type Annotations struct {
Records []Annotation `json:"records" required:"true"`
Total int `json:"total" required:"true"`
HistoryAvailableFrom int64 `json:"historyAvailableFrom" required:"true"`
}
type Annotation struct {
Kind AnnotationKind `json:"kind" required:"true"`
Spec any `json:"spec" required:"true"`
}
type PostableAnnotations struct {
Kind AnnotationKind `json:"kind" required:"true"`
Start int64 `json:"start" required:"true"`
End int64 `json:"end" required:"true"`
Signals []telemetrytypes.Signal `json:"signals"`
Filter *qbtypes.Filter `json:"filter"`
Limit int `json:"limit" required:"true"`
}
var (
_ jsonschema.OneOfExposer = Annotation{}
_ jsonschema.Preparer = Annotation{}
)
// annotationDeployment is the OpenAPI schema for an Annotation with kind=deployment.
type annotationDeployment struct {
Kind AnnotationKind `json:"kind" required:"true"`
Spec DeploymentSpec `json:"spec" required:"true"`
}
func (Annotation) JSONSchemaOneOf() []any {
return []any{
annotationDeployment{},
}
}
func (Annotation) PrepareJSONSchema(schema *jsonschema.Schema) error {
if schema.ExtraProperties == nil {
schema.ExtraProperties = map[string]any{}
}
schema.ExtraProperties["x-signoz-discriminator"] = map[string]any{
"propertyName": "kind",
"mapping": map[string]string{
AnnotationKindDeployment.StringValue(): "#/components/schemas/AnnotationtypesAnnotationDeployment",
},
}
return nil
}
func (req *PostableAnnotations) Validate() error {
if req == nil {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "request is nil")
}
if req.Kind != AnnotationKindDeployment {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid kind %q: only %q is supported", req.Kind.StringValue(), AnnotationKindDeployment.StringValue())
}
if req.Start <= 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid start time %d: start must be greater than 0", req.Start)
}
if req.End <= 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid end time %d: end must be greater than 0", req.End)
}
if req.Start >= req.End {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid time range: start (%d) must be less than end (%d)", req.Start, req.End)
}
for _, signal := range req.Signals {
if !slices.Contains(ValidSignals, signal) {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid signal %q: must be one of %q, %q", signal.StringValue(), telemetrytypes.SignalTraces.StringValue(), telemetrytypes.SignalLogs.StringValue())
}
}
if req.Limit < 1 || req.Limit > 5000 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be between 1 and 5000")
}
return nil
}
// UnmarshalJSON validates input immediately after decoding.
func (req *PostableAnnotations) UnmarshalJSON(data []byte) error {
type raw PostableAnnotations
var decoded raw
if err := json.Unmarshal(data, &decoded); err != nil {
return err
}
*req = PostableAnnotations(decoded)
return req.Validate()
}

View File

@@ -0,0 +1,159 @@
package annotationtypes
import (
"encoding/json"
"testing"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestPostableAnnotations_Validate(t *testing.T) {
testCases := []struct {
name string
req *PostableAnnotations
wantErr bool
}{
{
name: "DeploymentKind_BothSignals_Valid",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Signals: []telemetrytypes.Signal{telemetrytypes.SignalTraces, telemetrytypes.SignalLogs}, Limit: 100},
wantErr: false,
},
{
name: "NoSignals_Valid",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Limit: 1},
wantErr: false,
},
{
name: "NilRequest_Rejected",
req: nil,
wantErr: true,
},
{
name: "UnknownKind_Rejected",
req: &PostableAnnotations{Kind: AnnotationKind{valuer.NewString("incident")}, Start: 1000, End: 2000, Limit: 100},
wantErr: true,
},
{
name: "EmptyKind_Rejected",
req: &PostableAnnotations{Start: 1000, End: 2000, Limit: 100},
wantErr: true,
},
{
name: "StartZero_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 0, End: 2000, Limit: 100},
wantErr: true,
},
{
name: "EndZero_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 0, Limit: 100},
wantErr: true,
},
{
name: "StartAfterEnd_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 3000, End: 2000, Limit: 100},
wantErr: true,
},
{
name: "StartEqualsEnd_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 2000, End: 2000, Limit: 100},
wantErr: true,
},
{
name: "MetricsSignal_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Signals: []telemetrytypes.Signal{telemetrytypes.SignalMetrics}, Limit: 100},
wantErr: true,
},
{
name: "LimitZero_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Limit: 0},
wantErr: true,
},
{
name: "LimitNegative_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Limit: -5},
wantErr: true,
},
{
name: "LimitAboveMax_Rejected",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Limit: 5001},
wantErr: true,
},
{
name: "LimitAtMax_Valid",
req: &PostableAnnotations{Kind: AnnotationKindDeployment, Start: 1000, End: 2000, Limit: 5000},
wantErr: false,
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
err := testCase.req.Validate()
if !testCase.wantErr {
assert.NoError(t, err)
return
}
require.Error(t, err)
assert.True(t, errors.Ast(err, errors.TypeInvalidInput))
})
}
}
func TestPostableAnnotations_UnmarshalJSON(t *testing.T) {
testCases := []struct {
name string
body string
want PostableAnnotations
wantErr bool
}{
{
name: "FullBody_Decoded",
body: `{"kind":"deployment","start":1000,"end":2000,"signals":["traces"],"filter":{"expression":"service.name = 'cart'"},"limit":50}`,
want: PostableAnnotations{
Kind: AnnotationKindDeployment,
Start: 1000,
End: 2000,
Signals: []telemetrytypes.Signal{telemetrytypes.SignalTraces},
Limit: 50,
},
wantErr: false,
},
{
name: "InvalidKind_ValidatedOnDecode",
body: `{"kind":"release","start":1000,"end":2000,"limit":50}`,
wantErr: true,
},
{
name: "MissingLimit_ValidatedOnDecode",
body: `{"kind":"deployment","start":1000,"end":2000}`,
wantErr: true,
},
{
name: "MalformedJSON_Rejected",
body: `{"kind":"deployment",`,
wantErr: true,
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
var got PostableAnnotations
err := json.Unmarshal([]byte(testCase.body), &got)
if testCase.wantErr {
assert.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, testCase.want.Kind, got.Kind)
assert.Equal(t, testCase.want.Start, got.Start)
assert.Equal(t, testCase.want.End, got.End)
assert.Equal(t, testCase.want.Signals, got.Signals)
assert.Equal(t, testCase.want.Limit, got.Limit)
require.NotNil(t, got.Filter)
assert.Equal(t, "service.name = 'cart'", got.Filter.Expression)
})
}
}

View File

@@ -0,0 +1,140 @@
package annotationtypes
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"sort"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
var deploymentIdentityKeys = []string{"service.name", "deployment.environment", "deployment.environment.name", "service.version"}
type DeploymentSpec struct {
ID string `json:"id" required:"true"`
Signal telemetrytypes.Signal `json:"signal" required:"true"`
Service string `json:"service" required:"true"`
Environment string `json:"environment" required:"true"`
Version string `json:"version" required:"true"`
FirstSeen int64 `json:"firstSeen" required:"true"`
TraceID string `json:"traceId" required:"true"`
SpanID string `json:"spanId" required:"true"`
LogID string `json:"logId" required:"true"`
ResourceAttributes map[string]string `json:"resourceAttributes" required:"true"`
}
// DeploymentRow is the storage shape one marker row is scanned into; FirstSeen is epoch ms.
type DeploymentRow struct {
Signal telemetrytypes.Signal
Service string
Environment string
Version string
FirstSeen int64
TraceID string
SpanID string
LogID string
Labels string
}
func NewDeploymentSpecFromRow(row DeploymentRow) (DeploymentSpec, error) {
resourceAttributes := map[string]string{}
if row.Labels != "" {
if err := json.Unmarshal([]byte(row.Labels), &resourceAttributes); err != nil {
return DeploymentSpec{}, errors.WrapInternalf(err, errors.CodeInternal, "failed to parse resource labels for %s/%s/%s", row.Service, row.Environment, row.Version)
}
}
for _, key := range deploymentIdentityKeys {
delete(resourceAttributes, key)
}
return DeploymentSpec{
ID: NewDeploymentID(row.Service, row.Environment, row.Version),
Signal: row.Signal,
Service: row.Service,
Environment: row.Environment,
Version: row.Version,
FirstSeen: row.FirstSeen,
TraceID: row.TraceID,
SpanID: row.SpanID,
LogID: row.LogID,
ResourceAttributes: resourceAttributes,
}, nil
}
func NewDeploymentSpecsFromRows(rows []DeploymentRow) ([]DeploymentSpec, error) {
specs := make([]DeploymentSpec, 0, len(rows))
for _, row := range rows {
spec, err := NewDeploymentSpecFromRow(row)
if err != nil {
return nil, err
}
specs = append(specs, spec)
}
return specs, nil
}
func NewDeploymentID(service, environment, version string) string {
sum := sha256.Sum256([]byte(strings.Join([]string{service, environment, version}, "\x00")))
return "dep_" + hex.EncodeToString(sum[:8])
}
// MergeDeployments keeps the earliest FirstSeen per (service, environment, version); on a tie the first input wins.
func MergeDeployments(specs ...[]DeploymentSpec) []DeploymentSpec {
byTuple := map[string]DeploymentSpec{}
for _, list := range specs {
for _, spec := range list {
key := strings.Join([]string{spec.Service, spec.Environment, spec.Version}, "\x00")
existing, ok := byTuple[key]
if !ok || spec.FirstSeen < existing.FirstSeen {
byTuple[key] = spec
}
}
}
merged := make([]DeploymentSpec, 0, len(byTuple))
for _, spec := range byTuple {
merged = append(merged, spec)
}
sort.Slice(merged, func(i, j int) bool {
if merged[i].FirstSeen != merged[j].FirstSeen {
return merged[i].FirstSeen < merged[j].FirstSeen
}
if merged[i].Service != merged[j].Service {
return merged[i].Service < merged[j].Service
}
if merged[i].Environment != merged[j].Environment {
return merged[i].Environment < merged[j].Environment
}
return merged[i].Version < merged[j].Version
})
return merged
}
// EarliestHistoryEdge returns the smallest non-zero edge, 0 when no signal has history.
func EarliestHistoryEdge(edges ...int64) int64 {
var earliest int64
for _, edge := range edges {
if edge != 0 && (earliest == 0 || edge < earliest) {
earliest = edge
}
}
return earliest
}
func NewAnnotationsFromDeployments(specs []DeploymentSpec, total int, historyAvailableFrom int64) *Annotations {
records := make([]Annotation, 0, len(specs))
for _, spec := range specs {
records = append(records, Annotation{Kind: AnnotationKindDeployment, Spec: spec})
}
return &Annotations{Records: records, Total: total, HistoryAvailableFrom: historyAvailableFrom}
}
func ApplyLimit(specs []DeploymentSpec, limit int) ([]DeploymentSpec, int) {
if limit >= len(specs) {
return specs, len(specs)
}
return specs[:limit], len(specs)
}

View File

@@ -0,0 +1,204 @@
package annotationtypes
import (
"testing"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestNewDeploymentID(t *testing.T) {
testCases := []struct {
name string
service string
env string
version string
wantEq bool
}{
{name: "SameTuple_SameID", service: "checkout", env: "prod", version: "1.2.0", wantEq: true},
{name: "DifferentVersion_DifferentID", service: "checkout", env: "prod", version: "1.2.1", wantEq: false},
{name: "DifferentEnv_DifferentID", service: "checkout", env: "staging", version: "1.2.0", wantEq: false},
{name: "DifferentService_DifferentID", service: "cart", env: "prod", version: "1.2.0", wantEq: false},
}
base := NewDeploymentID("checkout", "prod", "1.2.0")
require.Len(t, base, len("dep_")+16)
assert.True(t, len(base) > 4 && base[:4] == "dep_")
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
got := NewDeploymentID(testCase.service, testCase.env, testCase.version)
assert.Equal(t, testCase.wantEq, got == base)
})
}
}
func TestNewDeploymentSpecFromRow(t *testing.T) {
testCases := []struct {
name string
row DeploymentRow
wantAttrs map[string]string
wantErr bool
}{
{
name: "IdentityKeysStripped_OthersKept",
row: DeploymentRow{
Signal: telemetrytypes.SignalTraces, Service: "checkout", Environment: "prod", Version: "1.2.0", FirstSeen: 1700000000123,
TraceID: "abc", SpanID: "def",
Labels: `{"service.name":"checkout","deployment.environment":"prod","deployment.environment.name":"prod","service.version":"1.2.0","k8s.namespace.name":"shop","host.name":"node-1"}`,
},
wantAttrs: map[string]string{"k8s.namespace.name": "shop", "host.name": "node-1"},
},
{
name: "EmptyLabels_EmptyMap",
row: DeploymentRow{Signal: telemetrytypes.SignalLogs, Service: "cart", Environment: "prod", Version: "2.0.0", LogID: "log-1"},
wantAttrs: map[string]string{},
},
{
name: "MalformedLabels_Error",
row: DeploymentRow{Service: "cart", Environment: "prod", Version: "2.0.0", Labels: `{"service.name":`},
wantErr: true,
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
spec, err := NewDeploymentSpecFromRow(testCase.row)
if testCase.wantErr {
assert.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, NewDeploymentID(testCase.row.Service, testCase.row.Environment, testCase.row.Version), spec.ID)
assert.Equal(t, testCase.row.Signal, spec.Signal)
assert.Equal(t, testCase.row.FirstSeen, spec.FirstSeen)
assert.Equal(t, testCase.row.TraceID, spec.TraceID)
assert.Equal(t, testCase.row.SpanID, spec.SpanID)
assert.Equal(t, testCase.row.LogID, spec.LogID)
assert.Equal(t, testCase.wantAttrs, spec.ResourceAttributes)
})
}
}
func TestMergeDeployments(t *testing.T) {
traces := []DeploymentSpec{
{Service: "checkout", Environment: "prod", Version: "1.2.0", FirstSeen: 200, Signal: telemetrytypes.SignalTraces, TraceID: "t1"},
{Service: "cart", Environment: "prod", Version: "3.0.0", FirstSeen: 100, Signal: telemetrytypes.SignalTraces, TraceID: "t2"},
{Service: "api", Environment: "prod", Version: "9.0.0", FirstSeen: 300, Signal: telemetrytypes.SignalTraces, TraceID: "t3"},
}
logs := []DeploymentSpec{
{Service: "checkout", Environment: "prod", Version: "1.2.0", FirstSeen: 150, Signal: telemetrytypes.SignalLogs, LogID: "l1"},
{Service: "cart", Environment: "prod", Version: "3.0.0", FirstSeen: 100, Signal: telemetrytypes.SignalLogs, LogID: "l2"},
{Service: "worker", Environment: "prod", Version: "0.1.0", FirstSeen: 300, Signal: telemetrytypes.SignalLogs, LogID: "l3"},
}
testCases := []struct {
name string
inputs [][]DeploymentSpec
wantOrder []string
wantSignalOf map[string]telemetrytypes.Signal
}{
{
name: "EarliestWins_TieKeepsFirstInput_SortedByFirstSeenThenService",
inputs: [][]DeploymentSpec{traces, logs},
wantOrder: []string{"cart", "checkout", "api", "worker"},
wantSignalOf: map[string]telemetrytypes.Signal{
"cart": telemetrytypes.SignalTraces,
"checkout": telemetrytypes.SignalLogs,
"api": telemetrytypes.SignalTraces,
"worker": telemetrytypes.SignalLogs,
},
},
{
name: "SingleInput_Sorted",
inputs: [][]DeploymentSpec{traces},
wantOrder: []string{"cart", "checkout", "api"},
wantSignalOf: map[string]telemetrytypes.Signal{"cart": telemetrytypes.SignalTraces, "checkout": telemetrytypes.SignalTraces, "api": telemetrytypes.SignalTraces},
},
{
name: "NoInput_Empty",
inputs: nil,
wantOrder: []string{},
wantSignalOf: map[string]telemetrytypes.Signal{},
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
merged := MergeDeployments(testCase.inputs...)
require.Len(t, merged, len(testCase.wantOrder))
for i, spec := range merged {
assert.Equal(t, testCase.wantOrder[i], spec.Service)
assert.Equal(t, testCase.wantSignalOf[spec.Service], spec.Signal)
}
})
}
}
func TestApplyLimit(t *testing.T) {
specs := []DeploymentSpec{{Service: "a"}, {Service: "b"}, {Service: "c"}}
testCases := []struct {
name string
limit int
wantLen int
wantTotal int
}{
{name: "LimitBelowCount_Truncated", limit: 2, wantLen: 2, wantTotal: 3},
{name: "LimitEqualsCount_All", limit: 3, wantLen: 3, wantTotal: 3},
{name: "LimitAboveCount_All", limit: 100, wantLen: 3, wantTotal: 3},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
page, total := ApplyLimit(specs, testCase.limit)
assert.Len(t, page, testCase.wantLen)
assert.Equal(t, testCase.wantTotal, total)
})
}
}
func TestEarliestHistoryEdge(t *testing.T) {
testCases := []struct {
name string
edges []int64
want int64
}{
{name: "BothZero_Zero", edges: []int64{0, 0}, want: 0},
{name: "OneZero_OtherWins", edges: []int64{0, 500}, want: 500},
{name: "BothSet_Smallest", edges: []int64{900, 500}, want: 500},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
assert.Equal(t, testCase.want, EarliestHistoryEdge(testCase.edges...))
})
}
}
func TestNewDeploymentSpecsFromRows(t *testing.T) {
specs, err := NewDeploymentSpecsFromRows([]DeploymentRow{
{Service: "a", Environment: "prod", Version: "1", Labels: `{"service.name":"a","host.name":"h1"}`},
{Service: "b", Environment: "prod", Version: "2"},
})
require.NoError(t, err)
require.Len(t, specs, 2)
assert.Equal(t, map[string]string{"host.name": "h1"}, specs[0].ResourceAttributes)
assert.Equal(t, map[string]string{}, specs[1].ResourceAttributes)
_, err = NewDeploymentSpecsFromRows([]DeploymentRow{{Labels: `{`}})
assert.Error(t, err)
}
func TestNewAnnotationsFromDeployments(t *testing.T) {
got := NewAnnotationsFromDeployments([]DeploymentSpec{{Service: "a"}}, 7, 1234)
require.Len(t, got.Records, 1)
assert.Equal(t, AnnotationKindDeployment, got.Records[0].Kind)
assert.Equal(t, 7, got.Total)
assert.Equal(t, int64(1234), got.HistoryAvailableFrom)
empty := NewAnnotationsFromDeployments(nil, 0, 0)
assert.NotNil(t, empty.Records)
assert.Len(t, empty.Records, 0)
}

View File

@@ -0,0 +1,2 @@
{ "timestamp": "2026-01-01T01:00:00.000000Z", "resources": { "service.name": "checkout", "deployment.environment": "prod", "service.version": "1.0.0" }, "attributes": {}, "body": "checkout 1.0.0 started", "severity_text": "INFO" }
{ "timestamp": "2026-01-01T02:39:30.000000Z", "resources": { "service.name": "checkout", "deployment.environment": "prod", "service.version": "1.1.0" }, "attributes": {}, "body": "checkout 1.1.0 started", "severity_text": "INFO" }

View File

@@ -0,0 +1,2 @@
{"timestamp": "2026-01-01T00:00:00.000000Z", "duration": "PT1.0S", "trace_id": "b0000000000000000000000000000001", "span_id": "b000000000000001", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.0.0", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T02:40:00.000000Z", "duration": "PT1.2S", "trace_id": "b0000000000000000000000000000002", "span_id": "b000000000000002", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.1.0", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/checkout"}}

View File

@@ -0,0 +1,4 @@
{ "timestamp": "2026-01-01T00:00:00.000000Z", "resources": { "service.name": "checkout", "deployment.environment": "prod", "service.version": "1.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a" }, "attributes": { "code.file": "checkout.py" }, "body": "checkout 1.0.0 started", "severity_text": "INFO" }
{ "timestamp": "2026-01-01T02:39:30.000000Z", "resources": { "service.name": "checkout", "deployment.environment": "prod", "service.version": "1.1.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a" }, "attributes": { "code.file": "checkout.py" }, "body": "checkout 1.1.0 started", "severity_text": "INFO" }
{ "timestamp": "2026-01-01T02:44:00.000000Z", "resources": { "service.name": "worker", "deployment.environment": "prod", "service.version": "3.0.0", "k8s.namespace.name": "jobs", "custom.team": "payments", "signoz.workspace.key.id": "key-b" }, "attributes": { "code.file": "worker.py" }, "body": "worker 3.0.0 started", "severity_text": "INFO" }
{ "timestamp": "2026-01-01T02:47:00.000000Z", "resources": { "service.name": "api", "deployment.environment": "prod", "service.version": "6.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-b" }, "attributes": { "code.file": "api.py" }, "body": "api 6.0.0 started", "severity_text": "INFO" }

View File

@@ -0,0 +1,11 @@
{"timestamp": "2026-01-01T00:00:00.000000Z", "duration": "PT1.0S", "trace_id": "a0000000000000000000000000000001", "span_id": "a000000000000001", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T01:00:00.000000Z", "duration": "PT1.2S", "trace_id": "a0000000000000000000000000000002", "span_id": "a000000000000002", "parent_span_id": "", "name": "GET /legacy", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "legacy", "deployment.environment": "prod", "service.version": "0.5.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-24", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/legacy"}}
{"timestamp": "2026-01-01T01:40:00.000000Z", "duration": "PT1.4S", "trace_id": "a0000000000000000000000000000003", "span_id": "a000000000000003", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T02:40:00.000000Z", "duration": "PT1.0S", "trace_id": "a0000000000000000000000000000004", "span_id": "a000000000000004", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.1.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T02:42:00.000000Z", "duration": "PT1.2S", "trace_id": "a0000000000000000000000000000005", "span_id": "a000000000000005", "parent_span_id": "", "name": "GET /cart", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "cart", "deployment.environment": "prod", "service.version": "2.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-b", "host.name": "ip-10-0-1-25", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/cart"}}
{"timestamp": "2026-01-01T02:44:00.000000Z", "duration": "PT1.4S", "trace_id": "a0000000000000000000000000000006", "span_id": "a000000000000006", "parent_span_id": "", "name": "run job", "kind": 1, "status_code": 1, "status_message": "", "resources": {"service.name": "worker", "deployment.environment": "prod", "service.version": "3.0.0", "k8s.namespace.name": "jobs", "custom.team": "payments", "signoz.workspace.key.id": "key-b", "host.name": "ip-10-0-2-10", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "job"}}
{"timestamp": "2026-01-01T02:46:00.000000Z", "duration": "PT1.0S", "trace_id": "a0000000000000000000000000000007", "span_id": "a000000000000007", "parent_span_id": "", "name": "GET /tie", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "tieA", "deployment.environment": "prod", "service.version": "4.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-26", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/tie"}}
{"timestamp": "2026-01-01T02:46:00.000000Z", "duration": "PT1.2S", "trace_id": "a0000000000000000000000000000008", "span_id": "a000000000000008", "parent_span_id": "", "name": "GET /tie", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "tieB", "deployment.environment": "prod", "service.version": "4.0.0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-27", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/tie"}}
{"timestamp": "2026-01-01T02:48:00.000000Z", "duration": "PT1.4S", "trace_id": "a0000000000000000000000000000009", "span_id": "a000000000000009", "parent_span_id": "", "name": "GET /special", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "sp`ec'ial", "deployment.environment": "pr$0d", "service.version": "5.0.0$0", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-28", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/special"}}
{"timestamp": "2026-01-01T02:50:00.000000Z", "duration": "PT1.0S", "trace_id": "a000000000000000000000000000000a", "span_id": "a00000000000000a", "parent_span_id": "", "name": "GET /noversion", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "noversion", "deployment.environment": "prod", "service.version": "", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-29", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/noversion"}}
{"timestamp": "2026-01-01T02:51:00.000000Z", "duration": "PT1.0S", "trace_id": "a000000000000000000000000000000b", "span_id": "a00000000000000b", "parent_span_id": "", "name": "GET /noservice", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "", "deployment.environment": "prod", "service.version": "9.9.9", "k8s.namespace.name": "shop", "custom.team": "payments", "signoz.workspace.key.id": "key-a", "host.name": "ip-10-0-1-31", "os.type": "linux"}, "attributes": {"net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "GET", "http.response.status_code": "200", "http.request.path": "/noservice"}}

View File

@@ -0,0 +1,4 @@
{"timestamp": "2026-01-01T00:00:00.000000Z", "duration": "PT1.0S", "trace_id": "c0000000000000000000000000000001", "span_id": "c000000000000001", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.0.0", "k8s.namespace.name": "shop", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"http.request.method": "GET", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T02:40:00.000000Z", "duration": "PT1.0S", "trace_id": "c0000000000000000000000000000002", "span_id": "c000000000000002", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment": "prod", "service.version": "1.1.0", "k8s.namespace.name": "shop", "host.name": "ip-10-0-1-23", "os.type": "linux"}, "attributes": {"http.request.method": "GET", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T02:41:00.000000Z", "duration": "PT1.0S", "trace_id": "c0000000000000000000000000000003", "span_id": "c000000000000003", "parent_span_id": "", "name": "GET /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "checkout", "deployment.environment.name": "staging", "service.version": "1.1.0", "k8s.namespace.name": "shop-staging", "host.name": "ip-10-0-9-23", "os.type": "linux"}, "attributes": {"http.request.method": "GET", "http.request.path": "/checkout"}}
{"timestamp": "2026-01-01T02:42:00.000000Z", "duration": "PT1.0S", "trace_id": "c0000000000000000000000000000004", "span_id": "c000000000000004", "parent_span_id": "", "name": "POST /invoice", "kind": 2, "status_code": 1, "status_message": "", "resources": {"service.name": "billing", "deployment.environment.name": "prod", "deployment.environment": "legacy-prod", "service.version": "7.0.0", "k8s.namespace.name": "shop", "host.name": "ip-10-0-1-30", "os.type": "linux"}, "attributes": {"http.request.method": "POST", "http.request.path": "/invoice"}}

View File

@@ -0,0 +1,461 @@
import re
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
import requests
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.fs import get_testdata_file_path
from fixtures.logs import Logs
from fixtures.traces import Traces
ENDPOINT = "/api/v1/annotations"
IDENTITY_KEYS = ("service.name", "deployment.environment", "deployment.environment.name", "service.version")
# load_from_file shifts the earliest dataset row (the baseline tuple) to now - BASELINE_OFFSET.
BASELINE_OFFSET = timedelta(hours=3)
ALL_MARKERS = {("checkout", "1.1.0"), ("cart", "2.0.0"), ("worker", "3.0.0"), ("tieA", "4.0.0"), ("tieB", "4.0.0"), ("api", "6.0.0"), ("sp`ec'ial", "5.0.0$0")}
def test_deployments_empty_tables(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces, # pylint: disable=unused-argument
insert_logs, # pylint: disable=unused-argument
) -> None:
"""No telemetry at all: 200 with no records and no history edge."""
now = datetime.now(tz=UTC).replace(microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50},
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
assert response.json()["data"] == {"records": [], "total": 0, "historyAvailableFrom": 0}
def test_deployments_accuracy(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
insert_logs,
) -> None:
"""Seed both signals; assert the full record contract, exact first-seen times,
deep-link ids, stripped identity keys, ordering, merge winners and the history edge."""
now = datetime.now(tz=UTC).replace(microsecond=0)
traces = Traces.load_from_file(get_testdata_file_path("annotations/traces.jsonl"), base_time=now - BASELINE_OFFSET)
logs = Logs.load_from_file(get_testdata_file_path("annotations/logs.jsonl"), base_time=now - BASELINE_OFFSET)
insert_traces(traces)
insert_logs(logs)
span_of: dict[tuple[str, str], Traces] = {}
for span in sorted(traces, key=lambda span: span.timestamp):
span_of.setdefault((span.resources_string["service.name"], span.resources_string["service.version"]), span)
log_of: dict[tuple[str, str], Logs] = {}
for log in sorted(logs, key=lambda log: log.timestamp):
log_of.setdefault((log.resources_string["service.name"], log.resources_string["service.version"]), log)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50},
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
# Ordered by firstSeen then service. checkout 1.1.0: logs 30s before traces, logs wins.
# worker 3.0.0: same timestamp on both signals, traces wins. tieA/tieB: same timestamp.
# checkout 1.0.0 (oldest bucket), legacy 0.5.0 (before start), noversion (empty version), noservice (empty service) are not markers.
expected = [
("checkout", "1.1.0", "logs"),
("cart", "2.0.0", "traces"),
("worker", "3.0.0", "traces"),
("tieA", "4.0.0", "traces"),
("tieB", "4.0.0", "traces"),
("api", "6.0.0", "logs"),
("sp`ec'ial", "5.0.0$0", "traces"),
]
assert [(r["spec"]["service"], r["spec"]["version"], r["spec"]["signal"]) for r in data["records"]] == expected
assert data["total"] == len(expected)
assert data["historyAvailableFrom"] == (int((now - BASELINE_OFFSET).timestamp()) // 1800 * 1800 + 1800) * 1000
for record in data["records"]:
assert record["kind"] == "deployment"
spec = record["spec"]
for field in ("id", "signal", "service", "environment", "version", "firstSeen", "traceId", "spanId", "logId", "resourceAttributes"):
assert field in spec, f"missing {field} in {spec!r}"
assert re.fullmatch(r"dep_[0-9a-f]{16}", spec["id"])
key = (spec["service"], spec["version"])
if spec["signal"] == "traces":
span = span_of[key]
assert spec["firstSeen"] == int(span.timestamp.timestamp() * 1000)
assert (spec["traceId"], spec["spanId"], spec["logId"]) == (span.trace_id, span.span_id, "")
resources = span.resources_string
else:
log = log_of[key]
assert spec["firstSeen"] == int(log.timestamp) // 1_000_000
assert (spec["traceId"], spec["spanId"], spec["logId"]) == ("", "", log.id)
resources = log.resources_string
assert spec["environment"] == resources["deployment.environment"]
assert spec["resourceAttributes"] == {k: v for k, v in resources.items() if k not in IDENTITY_KEYS}
assert len({r["spec"]["id"] for r in data["records"]}) == len(expected)
@pytest.mark.parametrize(
("start_anchor", "start_offset", "end_anchor", "end_offset", "expected"),
[
pytest.param("cart", 0, "now", 0, ALL_MARKERS - {("checkout", "1.1.0")}, id="start_at_first_row_inclusive"),
pytest.param("cart", 1, "now", 0, ALL_MARKERS - {("checkout", "1.1.0"), ("cart", "2.0.0")}, id="start_after_first_row"),
pytest.param("baseline", 7200, "cart", 0, {("checkout", "1.1.0"), ("cart", "2.0.0")}, id="end_at_first_row_inclusive"),
pytest.param("baseline", 7200, "cart", -1, {("checkout", "1.1.0")}, id="end_before_first_row"),
pytest.param("baseline", -60, "baseline", 60, set(), id="oldest_bucket_only"),
pytest.param("baseline", -7200, "baseline", -3600, set(), id="window_before_data"),
pytest.param("now", -300, "now", 0, set(), id="window_after_data"),
],
)
def test_deployments_exclusions( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
insert_logs,
start_anchor: str,
start_offset: int,
end_anchor: str,
end_offset: int,
expected: set[tuple[str, str]],
) -> None:
"""Window bounds bind exactly on first_seen (inclusive both ends). Rows in the oldest
bucket, before start, or outside the window never produce markers, and the history
edge is reported even when nothing matches."""
now = datetime.now(tz=UTC).replace(microsecond=0)
traces = Traces.load_from_file(get_testdata_file_path("annotations/traces.jsonl"), base_time=now - BASELINE_OFFSET)
logs = Logs.load_from_file(get_testdata_file_path("annotations/logs.jsonl"), base_time=now - BASELINE_OFFSET)
insert_traces(traces)
insert_logs(logs)
anchors = {"cart": next(span.timestamp for span in traces if span.resources_string["service.name"] == "cart"), "baseline": now - BASELINE_OFFSET, "now": now}
start = anchors[start_anchor] + timedelta(seconds=start_offset)
end = anchors[end_anchor] + timedelta(seconds=end_offset)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={"kind": "deployment", "start": int(start.timestamp() * 1000), "end": int(end.timestamp() * 1000), "limit": 50},
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
assert {(r["spec"]["service"], r["spec"]["version"]) for r in data["records"]} == expected
assert data["total"] == len(expected)
assert data["historyAvailableFrom"] == (int((now - BASELINE_OFFSET).timestamp()) // 1800 * 1800 + 1800) * 1000
@pytest.mark.parametrize(
("signals", "expected"),
[
pytest.param(None, {("checkout", "1.1.0", "logs"), ("cart", "2.0.0", "traces"), ("worker", "3.0.0", "traces"), ("tieA", "4.0.0", "traces"), ("tieB", "4.0.0", "traces"), ("api", "6.0.0", "logs"), ("sp`ec'ial", "5.0.0$0", "traces")}, id="unset_both"),
pytest.param([], {("checkout", "1.1.0", "logs"), ("cart", "2.0.0", "traces"), ("worker", "3.0.0", "traces"), ("tieA", "4.0.0", "traces"), ("tieB", "4.0.0", "traces"), ("api", "6.0.0", "logs"), ("sp`ec'ial", "5.0.0$0", "traces")}, id="empty_list_both"),
pytest.param(["traces"], {("checkout", "1.1.0", "traces"), ("cart", "2.0.0", "traces"), ("worker", "3.0.0", "traces"), ("tieA", "4.0.0", "traces"), ("tieB", "4.0.0", "traces"), ("sp`ec'ial", "5.0.0$0", "traces")}, id="traces_only"),
pytest.param(["logs"], {("checkout", "1.1.0", "logs"), ("worker", "3.0.0", "logs"), ("api", "6.0.0", "logs")}, id="logs_only"),
pytest.param(["logs", "traces"], {("checkout", "1.1.0", "logs"), ("cart", "2.0.0", "traces"), ("worker", "3.0.0", "traces"), ("tieA", "4.0.0", "traces"), ("tieB", "4.0.0", "traces"), ("api", "6.0.0", "logs"), ("sp`ec'ial", "5.0.0$0", "traces")}, id="explicit_both"),
],
)
def test_deployments_signals( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
insert_logs,
signals: list[str] | None,
expected: set[tuple[str, str, str]],
) -> None:
"""Signal selection: a single signal only ever reports that signal and its ids; the
deployment id of a tuple is the same whichever signal reports it."""
now = datetime.now(tz=UTC).replace(microsecond=0)
traces = Traces.load_from_file(get_testdata_file_path("annotations/traces.jsonl"), base_time=now - BASELINE_OFFSET)
logs = Logs.load_from_file(get_testdata_file_path("annotations/logs.jsonl"), base_time=now - BASELINE_OFFSET)
insert_traces(traces)
insert_logs(logs)
body: dict = {"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50}
if signals is not None:
body["signals"] = signals
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(signoz.self.host_configs["8080"].get(ENDPOINT), headers={"authorization": f"Bearer {token}"}, json=body, timeout=5)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
assert {(r["spec"]["service"], r["spec"]["version"], r["spec"]["signal"]) for r in data["records"]} == expected
assert data["total"] == len(expected)
for record in data["records"]:
spec = record["spec"]
if spec["signal"] == "traces":
assert spec["traceId"] and spec["spanId"] and spec["logId"] == ""
else:
assert spec["logId"] and spec["traceId"] == "" and spec["spanId"] == ""
# Same tuple, other signal: same deployment id.
other = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={**body, "signals": ["traces"] if signals == ["logs"] else ["logs"]},
timeout=5,
)
assert other.status_code == HTTPStatus.OK, other.text
ids = {(r["spec"]["service"], r["spec"]["version"]): r["spec"]["id"] for r in data["records"]}
for record in other.json()["data"]["records"]:
key = (record["spec"]["service"], record["spec"]["version"])
if key in ids:
assert record["spec"]["id"] == ids[key]
@pytest.mark.parametrize(
("signals", "edge_offset"),
[
pytest.param(None, timedelta(hours=3), id="both_earliest_edge"),
pytest.param(["traces"], timedelta(hours=3), id="traces_edge"),
pytest.param(["logs"], timedelta(hours=2), id="logs_edge"),
],
)
def test_deployments_history_edge( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
insert_logs,
signals: list[str] | None,
edge_offset: timedelta,
) -> None:
"""historyAvailableFrom is the bucket after the oldest resource bucket of the queried
signals; with both signals it is the earlier of the two edges."""
now = datetime.now(tz=UTC).replace(microsecond=0)
insert_traces(Traces.load_from_file(get_testdata_file_path("annotations/edge_traces.jsonl"), base_time=now - timedelta(hours=3)))
insert_logs(Logs.load_from_file(get_testdata_file_path("annotations/edge_logs.jsonl"), base_time=now - timedelta(hours=2)))
body: dict = {"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50}
if signals is not None:
body["signals"] = signals
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(signoz.self.host_configs["8080"].get(ENDPOINT), headers={"authorization": f"Bearer {token}"}, json=body, timeout=5)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
assert data["historyAvailableFrom"] == (int((now - edge_offset).timestamp()) // 1800 * 1800 + 1800) * 1000
assert {(r["spec"]["service"], r["spec"]["version"]) for r in data["records"]} == {("checkout", "1.1.0")}
@pytest.mark.parametrize(
("families_on", "expression", "expected"),
[
pytest.param(True, "", {("checkout", "prod", "1.1.0"), ("checkout", "staging", "1.1.0"), ("billing", "prod", "7.0.0")}, id="flag_on_reads_both_spellings_current_first"),
pytest.param(True, "deployment.environment = 'staging'", {("checkout", "staging", "1.1.0")}, id="flag_on_filter_widened_to_new_key"),
pytest.param(False, "", {("checkout", "prod", "1.1.0"), ("checkout", "", "1.1.0"), ("billing", "legacy-prod", "7.0.0")}, id="flag_off_reads_legacy_key_only"),
pytest.param(False, "deployment.environment = 'staging'", set(), id="flag_off_filter_stays_literal"),
],
)
def test_deployments_environment_family( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
signoz_families_on: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
families_on: bool,
expression: str,
expected: set[tuple[str, str, str]],
) -> None:
"""With resolve_semconv_families on, the environment is read from
deployment.environment.name first, then deployment.environment, so a
resource emitting only the new key keeps its own tuple and id. With the
flag off the legacy key alone is read, as the filter path does."""
now = datetime.now(tz=UTC).replace(microsecond=0)
traces = Traces.load_from_file(get_testdata_file_path("annotations/traces_environment.jsonl"), base_time=now - BASELINE_OFFSET)
insert_traces(traces)
instance = signoz_families_on if families_on else signoz
body: dict = {"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50}
if expression:
body["filter"] = {"expression": expression}
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(instance.self.host_configs["8080"].get(ENDPOINT), headers={"authorization": f"Bearer {token}"}, json=body, timeout=5)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
assert {(r["spec"]["service"], r["spec"]["environment"], r["spec"]["version"]) for r in data["records"]} == expected
assert data["total"] == len(expected)
assert len({r["spec"]["id"] for r in data["records"]}) == len(expected)
for record in data["records"]:
assert "deployment.environment.name" not in record["spec"]["resourceAttributes"]
assert "deployment.environment" not in record["spec"]["resourceAttributes"]
@pytest.mark.parametrize(
("expression", "expected"),
[
pytest.param("k8s.namespace.name = 'jobs'", {("worker", "3.0.0")}, id="eq"),
pytest.param("k8s.namespace.name IN ('jobs', 'shop')", ALL_MARKERS, id="in"),
pytest.param("k8s.namespace.name NOT IN ('shop')", {("worker", "3.0.0")}, id="not_in"),
pytest.param("service.name CONTAINS 'tie'", {("tieA", "4.0.0"), ("tieB", "4.0.0")}, id="contains"),
pytest.param("service.name LIKE 'tie%'", {("tieA", "4.0.0"), ("tieB", "4.0.0")}, id="like"),
pytest.param("k8s.namespace.name = 'shop' AND service.name = 'cart'", {("cart", "2.0.0")}, id="and"),
pytest.param("service.name = 'cart' OR service.name = 'api'", {("cart", "2.0.0"), ("api", "6.0.0")}, id="or_across_signals"),
pytest.param("deployment.environment = 'pr$0d'", {("sp`ec'ial", "5.0.0$0")}, id="dollar_in_value"),
pytest.param('service.name = "sp`ec\'ial"', {("sp`ec'ial", "5.0.0$0")}, id="backtick_quote_in_value"),
pytest.param("custom.team = 'payments'", ALL_MARKERS, id="custom_resource_attribute"),
pytest.param("signoz.workspace.key.id = 'key-b'", {("cart", "2.0.0"), ("worker", "3.0.0"), ("api", "6.0.0")}, id="workspace_key"),
pytest.param("service.name = 'nope'", set(), id="no_match"),
pytest.param("k8s.namespace.namee = 'jobs'", ALL_MARKERS, id="unresolved_key_noop"),
pytest.param("http.request.method = 'GET'", ALL_MARKERS, id="span_attribute_noop"),
pytest.param("", ALL_MARKERS, id="empty_expression"),
],
)
def test_deployments_filter( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
insert_logs,
expression: str,
expected: set[tuple[str, str]],
) -> None:
"""Filters apply to resource attributes on both signals; keys that are not resource
attributes compile to nothing and leave the result unfiltered."""
now = datetime.now(tz=UTC).replace(microsecond=0)
insert_traces(Traces.load_from_file(get_testdata_file_path("annotations/traces.jsonl"), base_time=now - BASELINE_OFFSET))
insert_logs(Logs.load_from_file(get_testdata_file_path("annotations/logs.jsonl"), base_time=now - BASELINE_OFFSET))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50, "filter": {"expression": expression}},
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
assert {(r["spec"]["service"], r["spec"]["version"]) for r in data["records"]} == expected
assert data["total"] == len(expected)
@pytest.mark.parametrize(
"expression",
[
pytest.param("service.name =", id="trailing_op"),
pytest.param("(service.name = 'cart'", id="unclosed_paren"),
],
)
def test_deployments_filter_invalid(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
expression: str,
) -> None:
now = datetime.now(tz=UTC).replace(microsecond=0)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50, "filter": {"expression": expression}},
timeout=5,
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert "syntax error" in response.json()["error"]["message"]
@pytest.mark.parametrize(
("limit", "expected_len"),
[
pytest.param(1, 1, id="below_count"),
pytest.param(7, 7, id="equals_count"),
pytest.param(5000, 7, id="above_count"),
],
)
def test_deployments_limit( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
insert_traces,
insert_logs,
limit: int,
expected_len: int,
) -> None:
"""limit truncates the merged, ordered list; total always counts every marker."""
now = datetime.now(tz=UTC).replace(microsecond=0)
insert_traces(Traces.load_from_file(get_testdata_file_path("annotations/traces.jsonl"), base_time=now - BASELINE_OFFSET))
insert_logs(Logs.load_from_file(get_testdata_file_path("annotations/logs.jsonl"), base_time=now - BASELINE_OFFSET))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}"},
json={"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": limit},
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()["data"]
assert len(data["records"]) == expected_len
assert data["total"] == len(ALL_MARKERS)
assert (data["records"][0]["spec"]["service"], data["records"][0]["spec"]["version"]) == ("checkout", "1.1.0")
assert [r["spec"]["firstSeen"] for r in data["records"]] == sorted(r["spec"]["firstSeen"] for r in data["records"])
@pytest.mark.parametrize(
("overrides", "message"),
[
pytest.param({"kind": None}, "invalid kind", id="kind_missing"),
pytest.param({"kind": "release"}, "invalid kind", id="kind_unknown"),
pytest.param({"start": 0}, "start must be greater than 0", id="start_zero"),
pytest.param({"start": -1}, "start must be greater than 0", id="start_negative"),
pytest.param({"end": 0}, "end must be greater than 0", id="end_zero"),
pytest.param({"end": "same"}, "must be less than end", id="start_equals_end"),
pytest.param({"end": "before"}, "must be less than end", id="start_greater_than_end"),
pytest.param({"signals": ["metrics"]}, "invalid signal", id="signal_unknown"),
pytest.param({"limit": None}, "limit must be between", id="limit_missing"),
pytest.param({"limit": 0}, "limit must be between", id="limit_zero"),
pytest.param({"limit": -1}, "limit must be between", id="limit_negative"),
pytest.param({"limit": 5001}, "limit must be between", id="limit_too_large"),
],
)
def test_deployments_validation_errors(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
overrides: dict,
message: str,
) -> None:
now = datetime.now(tz=UTC).replace(microsecond=0)
body: dict = {"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50}
for key, value in overrides.items():
if value is None:
body.pop(key)
elif value == "same":
body[key] = body["start"]
elif value == "before":
body[key] = body["start"] - 1
else:
body[key] = value
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(signoz.self.host_configs["8080"].get(ENDPOINT), headers={"authorization": f"Bearer {token}"}, json=body, timeout=5)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert message in response.json()["error"]["message"]
def test_deployments_malformed_body(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token,
) -> None:
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
headers={"authorization": f"Bearer {token}", "content-type": "application/json"},
data='{"kind": "deployment", "start": ',
timeout=5,
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text

View File

@@ -0,0 +1,150 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
import requests
from wiremock.resources.mappings import Mapping
from fixtures import types
from fixtures.auth import (
USER_ADMIN_EMAIL,
USER_ADMIN_PASSWORD,
add_license,
change_user_role,
create_active_user,
)
from fixtures.fs import get_testdata_file_path
from fixtures.logs import Logs
from fixtures.role import transaction_group
from fixtures.traces import Traces
ENDPOINT = "/api/v1/annotations"
user_password = "password123Z$"
users = {
"metrics_only": ("annotations-metrics-only", "annotations-metrics-only@telemetry.test"),
"traces_only": ("annotations-traces-only", "annotations-traces-only@telemetry.test"),
"logs_only": ("annotations-logs-only", "annotations-logs-only@telemetry.test"),
"both_wildcard": ("annotations-both-wildcard", "annotations-both-wildcard@telemetry.test"),
"both_scoped": ("annotations-both-scoped", "annotations-both-scoped@telemetry.test"),
}
def test_setup(
signoz: types.SigNoz,
create_user_admin: types.Operation, # pylint: disable=unused-argument
make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
get_token: Callable[[str, str], str],
create_role: Callable[..., str],
) -> None:
add_license(signoz, make_http_mocks, get_token)
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
grants = {
"metrics_only": [transaction_group("read", "telemetryresource", "metrics", ["builder_query/*"])],
"traces_only": [transaction_group("read", "telemetryresource", "traces", ["builder_query/*"])],
"logs_only": [transaction_group("read", "telemetryresource", "logs", ["builder_query/*"])],
"both_wildcard": [
transaction_group("read", "telemetryresource", "traces", ["builder_query/*"]),
transaction_group("read", "telemetryresource", "logs", ["builder_query/*"]),
],
"both_scoped": [
transaction_group("read", "telemetryresource", "traces", ["builder_query/signoz.workspace.key.id/key-a"]),
transaction_group("read", "telemetryresource", "logs", ["builder_query/signoz.workspace.key.id/key-a"]),
],
}
for name, (role, email) in users.items():
create_role(admin_token, role, grants[name])
user = create_active_user(signoz, admin_token, email=email, role="signoz-viewer", password=user_password)
change_user_role(signoz, admin_token, user, "signoz-viewer", role)
@pytest.mark.parametrize(
("user", "expected"),
[
pytest.param("metrics_only", HTTPStatus.FORBIDDEN, id="metrics_only_denied"),
pytest.param("traces_only", HTTPStatus.FORBIDDEN, id="traces_only_denied"),
pytest.param("logs_only", HTTPStatus.FORBIDDEN, id="logs_only_denied"),
pytest.param("both_wildcard", HTTPStatus.OK, id="both_wildcard_allowed"),
pytest.param("both_scoped", HTTPStatus.FORBIDDEN, id="both_scoped_unfiltered_denied"),
pytest.param("admin", HTTPStatus.OK, id="admin_allowed"),
pytest.param(None, HTTPStatus.UNAUTHORIZED, id="anonymous"),
],
)
def test_list_route(
signoz: types.SigNoz,
get_token: Callable[[str, str], str],
user: str | None,
expected: HTTPStatus,
) -> None:
"""The route declares traces and logs read; a grant on only one signal is denied."""
now = datetime.now(tz=UTC)
headers = {}
if user == "admin":
headers["Authorization"] = f"Bearer {get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)}"
elif user is not None:
headers["Authorization"] = f"Bearer {get_token(users[user][1], user_password)}"
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
json={"kind": "deployment", "start": int((now - timedelta(minutes=10)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 10},
headers=headers,
timeout=10,
)
assert response.status_code == expected, response.text
@pytest.mark.parametrize("user", ["metrics_only", "traces_only", "logs_only"])
def test_denied_message_names_wildcard(
signoz: types.SigNoz,
get_token: Callable[[str, str], str],
user: str,
) -> None:
now = datetime.now(tz=UTC)
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
json={"kind": "deployment", "start": int((now - timedelta(minutes=10)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 10},
headers={"Authorization": f"Bearer {get_token(users[user][1], user_password)}"},
timeout=10,
)
assert response.status_code == HTTPStatus.FORBIDDEN, response.text
assert "builder_query/*" in response.text
@pytest.mark.parametrize(
("user", "expression", "expected", "expected_markers"),
[
pytest.param("both_scoped", "signoz.workspace.key.id = 'key-a'", HTTPStatus.OK, {("checkout", "1.1.0"), ("tieA", "4.0.0"), ("tieB", "4.0.0"), ("sp`ec'ial", "5.0.0$0")}, id="scoped_granted_key"),
pytest.param("both_scoped", "signoz.workspace.key.id = 'key-b'", HTTPStatus.FORBIDDEN, set(), id="scoped_other_key"),
pytest.param("both_scoped", "signoz.workspace.key.id IN ('key-a', 'key-b')", HTTPStatus.FORBIDDEN, set(), id="scoped_both_keys"),
pytest.param("both_scoped", "signoz.workspace.key.id = 'key-a' AND k8s.namespace.name = 'shop'", HTTPStatus.OK, {("checkout", "1.1.0"), ("tieA", "4.0.0"), ("tieB", "4.0.0"), ("sp`ec'ial", "5.0.0$0")}, id="scoped_granted_key_with_extra_filter"),
pytest.param("both_wildcard", None, HTTPStatus.OK, {("checkout", "1.1.0"), ("cart", "2.0.0"), ("worker", "3.0.0"), ("tieA", "4.0.0"), ("tieB", "4.0.0"), ("api", "6.0.0"), ("sp`ec'ial", "5.0.0$0")}, id="wildcard_unfiltered"),
],
)
def test_scoped_read( # pylint: disable=too-many-arguments,too-many-positional-arguments
signoz: types.SigNoz,
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
insert_logs: Callable[[list[Logs]], None],
user: str,
expression: str | None,
expected: HTTPStatus,
expected_markers: set[tuple[str, str]],
) -> None:
"""A key-scoped grant on both signals passes only when the filter pins that key, and
the response then carries only that key's markers from either signal."""
now = datetime.now(tz=UTC).replace(microsecond=0)
insert_traces(Traces.load_from_file(get_testdata_file_path("annotations/traces.jsonl"), base_time=now - timedelta(hours=3)))
insert_logs(Logs.load_from_file(get_testdata_file_path("annotations/logs.jsonl"), base_time=now - timedelta(hours=3)))
body: dict = {"kind": "deployment", "start": int((now - timedelta(hours=1)).timestamp() * 1000), "end": int(now.timestamp() * 1000), "limit": 50}
if expression:
body["filter"] = {"expression": expression}
response = requests.post(
signoz.self.host_configs["8080"].get(ENDPOINT),
json=body,
headers={"Authorization": f"Bearer {get_token(users[user][1], user_password)}"},
timeout=10,
)
assert response.status_code == expected, response.text
if expected == HTTPStatus.OK:
assert {(r["spec"]["service"], r["spec"]["version"]) for r in response.json()["data"]["records"]} == expected_markers

View File

@@ -0,0 +1,32 @@
import pytest
from testcontainers.core.container import Network
from fixtures import types
from fixtures.signoz import create_signoz
@pytest.fixture(name="signoz_families_on", scope="package")
def signoz_families_on(
network: Network,
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
"""Shares the sqlstore and clickhouse with the default instance, so the
same admin token and seeded rows work."""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-annotations-families-on",
env_overrides={
"SIGNOZ_FLAGGER_CONFIG_BOOLEAN_RESOLVE__SEMCONV__FAMILIES": True,
},
)