Compare commits

..

2 Commits

Author SHA1 Message Date
Yunus M
92f3f51eb6 chore: migrate .cursorrules to .cursor/rules/ format 2026-02-19 00:11:23 +05:30
Karan Balani
6aaea79b73 chore: add tests for unique index on email and org_id in users table (#10331)
Some checks failed
build-staging / prepare (push) Has been cancelled
build-staging / js-build (push) Has been cancelled
build-staging / go-build (push) Has been cancelled
build-staging / staging (push) Has been cancelled
Release Drafter / update_release_draft (push) Has been cancelled
* chore: add tests for unique index on email and org_id in users table

* chore: choose one status code

* chore: choose one status code

* chore: move unique index test into passwordauthn folder

* chore: moved to existing suite, remove register

* chore: better name for file and vars

* chore: fix var name

* chore: fix var name

* chore: fix var name
2026-02-18 19:05:50 +05:30
25 changed files with 968 additions and 3516 deletions

2
.gitignore vendored
View File

@@ -229,5 +229,3 @@ cython_debug/
pyrightconfig.json
# cursor files
frontend/.cursor/

View File

@@ -607,146 +607,6 @@ paths:
summary: Update auth domain
tags:
- authdomains
/api/v1/export_raw_data:
get:
deprecated: false
description: This endpoints allows simple query exporting raw data for traces
and logs
operationId: HandleExportRawDataGET
parameters:
- in: query
name: format
schema:
default: csv
enum:
- csv
- jsonl
type: string
- in: query
name: source
schema:
default: logs
enum:
- logs
- traces
type: string
- in: query
name: start
schema:
minimum: 0
type: integer
- in: query
name: end
schema:
minimum: 0
type: integer
- in: query
name: limit
schema:
default: 10000
maximum: 50000
minimum: 1
type: integer
- in: query
name: filter
schema:
type: string
- in: query
name: columns
schema:
items:
type: string
type: array
- in: query
name: order_by
schema:
type: string
responses:
"200":
content:
application/json:
schema:
type: string
description: OK
"400":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Bad Request
"500":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
summary: Export raw data
tags:
- logs
- traces
post:
deprecated: false
description: This endpoints allows complex query exporting raw data for traces
and logs
operationId: HandleExportRawDataPOST
parameters:
- in: query
name: format
schema:
default: csv
enum:
- csv
- jsonl
type: string
requestBody:
content:
application/json:
schema:
properties:
compositeQuery:
$ref: '#/components/schemas/Querybuildertypesv5CompositeQuery'
end:
minimum: 0
type: integer
formatOptions:
$ref: '#/components/schemas/Querybuildertypesv5FormatOptions'
noCache:
type: boolean
requestType:
type: string
schemaVersion:
type: string
start:
minimum: 0
type: integer
variables:
additionalProperties:
$ref: '#/components/schemas/Querybuildertypesv5VariableItem'
type: object
type: object
responses:
"200":
content:
application/json:
schema:
type: string
description: OK
"400":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Bad Request
"500":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
summary: Export raw data
tags:
- logs
- traces
/api/v1/fields/keys:
get:
deprecated: false
@@ -4452,14 +4312,6 @@ components:
type:
type: string
type: object
Querybuildertypesv5CompositeQuery:
properties:
queries:
items:
$ref: '#/components/schemas/Querybuildertypesv5QueryEnvelope'
nullable: true
type: array
type: object
Querybuildertypesv5ExecStats:
properties:
bytesScanned:
@@ -4482,13 +4334,6 @@ components:
expression:
type: string
type: object
Querybuildertypesv5FormatOptions:
properties:
fillGaps:
type: boolean
formatTableResultForUI:
type: boolean
type: object
Querybuildertypesv5OrderBy:
properties:
direction:
@@ -4520,12 +4365,6 @@ components:
nullable: true
type: array
type: object
Querybuildertypesv5QueryEnvelope:
properties:
spec: {}
type:
type: string
type: object
Querybuildertypesv5QueryRangeResponse:
properties:
data:
@@ -4553,12 +4392,6 @@ components:
message:
type: string
type: object
Querybuildertypesv5VariableItem:
properties:
type:
type: string
value: {}
type: object
RenderErrorResponse:
properties:
error:

View File

@@ -0,0 +1,74 @@
---
description: Core testing conventions - imports, rendering, MSW, interactions, queries
globs: **/*.test.{ts,tsx}
alwaysApply: false
---
# Testing Conventions
## Imports
Always import from the test harness, never directly from `@testing-library/react`:
```ts
import { render, screen, userEvent, waitFor } from 'tests/test-utils';
import { server, rest } from 'mocks-server/server';
```
## Router
Use the built-in router in `render`:
```ts
render(<Page />, undefined, { initialRoute: '/traces-explorer' });
```
Only mock `useLocation` / `useParams` if the test depends on their values.
## MSW
Global MSW server runs automatically. Override per-test:
```ts
server.use(
rest.get('*/api/v1/foo', (_req, res, ctx) =>
res(ctx.status(200), ctx.json({ ok: true })))
);
```
Keep large response fixtures in `mocks-server/__mockdata_`.
## Interactions
- Prefer `userEvent` for real user interactions (click, type, select, tab).
- Use `fireEvent` only for low-level events not covered by `userEvent` (e.g., scroll, resize). Wrap in `act(...)` if needed.
- Always `await` interactions:
```ts
const user = userEvent.setup({ pointerEventsCheck: 0 });
await user.click(screen.getByRole('button', { name: /save/i }));
```
## Timers
No global fake timers. Per-test only, for debounce/throttle:
```ts
jest.useFakeTimers();
const user = userEvent.setup({ advanceTimers: (ms) => jest.advanceTimersByTime(ms) });
await user.type(screen.getByRole('textbox'), 'query');
jest.advanceTimersByTime(400);
jest.useRealTimers();
```
## Queries
Prefer accessible queries: `getByRole` > `findByRole` > `getByLabelText` > visible text > `data-testid` (last resort).
## Anti-patterns
- Never import from `@testing-library/react` directly
- Never use global fake timers
- Never wrap `render` in `act(...)`
- Never mock infra dependencies locally (router, react-query)
- Limit to 3-5 focused tests per file

View File

@@ -0,0 +1,54 @@
---
description: When to use global vs local mocks in tests
globs: **/*.test.{ts,tsx}
alwaysApply: false
---
# Mock Strategy
## Use Global Mocks For
High-frequency dependencies (20+ test files):
- Core infrastructure: react-router-dom, react-query, antd
- Browser APIs: ResizeObserver, matchMedia, localStorage
- Utility libraries: date-fns, lodash
Available global mock files (from jest.config.ts):
- `uplot` -> `__mocks__/uplotMock.ts`
## Use Local Mocks For
- Business logic dependencies (API endpoints, custom hooks, domain components)
- Test-specific behavior (different data per test, error scenarios, loading states)
## Decision Tree
```
Used in 20+ test files?
YES -> Global mock
NO -> Business logic or test-specific?
YES -> Local mock
NO -> Consider global if usage grows
```
## Correct Usage
```ts
// Global mocks are already available - just import
import { useLocation } from 'react-router-dom';
// Local mocks for business logic
jest.mock('../api/tracesService', () => ({
getTraces: jest.fn(() => mockTracesData),
}));
```
## Anti-patterns
```ts
// Never re-mock globally mocked dependencies locally
jest.mock('react-router-dom', () => ({ ... }));
// Never put test-specific data in global mocks
jest.mock('../api/tracesService', () => ({ getTraces: jest.fn(() => specificTestData) }));
```

View File

@@ -0,0 +1,54 @@
---
description: TypeScript type safety requirements for Jest tests
globs: **/*.test.{ts,tsx}
alwaysApply: false
---
# TypeScript Type Safety in Tests
All Jest tests must be fully type-safe. Never use `any`.
## Mock Function Typing
```ts
// Use jest.mocked for module mocks
import useFoo from 'hooks/useFoo';
jest.mock('hooks/useFoo');
const mockUseFoo = jest.mocked(useFoo);
// Use jest.MockedFunction for standalone mocks
const mockFetch = jest.fn() as jest.MockedFunction<(id: number) => Promise<User>>;
```
## Mock Data
Define interfaces for all mock data:
```ts
const mockUser: User = { id: 1, name: 'John', email: 'john@example.com' };
const mockProps: ComponentProps = {
title: 'Test',
data: [mockUser],
onSelect: jest.fn() as jest.MockedFunction<(user: User) => void>,
};
```
## Hook Mocking Pattern
```ts
import useFoo from 'hooks/useFoo';
jest.mock('hooks/useFoo');
const mockUseFoo = jest.mocked(useFoo);
mockUseFoo.mockReturnValue(/* minimal shape */);
```
Prefer helpers (`rqSuccess`, `rqLoading`, `rqError`) for React Query results.
## Checklist
- All mock functions use `jest.MockedFunction<T>` or `jest.mocked()`
- All mock data has proper interfaces
- No `any` types in test files
- Component props are typed
- API response types are defined

View File

@@ -1,484 +0,0 @@
# Persona
You are an expert developer with deep knowledge of Jest, React Testing Library, MSW, and TypeScript, tasked with creating unit tests for this repository.
# Auto-detect TypeScript Usage
Check for TypeScript in the project through tsconfig.json or package.json dependencies.
Adjust syntax based on this detection.
# TypeScript Type Safety for Jest Tests
**CRITICAL**: All Jest tests MUST be fully type-safe with proper TypeScript types.
**Type Safety Requirements:**
- Use proper TypeScript interfaces for all mock data
- Type all Jest mock functions with `jest.MockedFunction<T>`
- Use generic types for React components and hooks
- Define proper return types for mock functions
- Use `as const` for literal types when needed
- Avoid `any` type use proper typing instead
# Unit Testing Focus
Focus on critical functionality (business logic, utility functions, component behavior)
Mock dependencies (API calls, external modules) before imports
Test multiple data scenarios (valid inputs, invalid inputs, edge cases)
Write maintainable tests with descriptive names grouped in describe blocks
# Global vs Local Mocks
**Use Global Mocks for:**
- High-frequency dependencies (20+ test files)
- Core infrastructure (react-router-dom, react-query, antd)
- Standard implementations across the app
- Browser APIs (ResizeObserver, matchMedia, localStorage)
- Utility libraries (date-fns, lodash)
**Use Local Mocks for:**
- Business logic dependencies (5-15 test files)
- Test-specific behavior (different data per test)
- API endpoints with specific responses
- Domain-specific components
- Error scenarios and edge cases
**Global Mock Files Available (from jest.config.ts):**
- `uplot` → `__mocks__/uplotMock.ts`
# Repo-specific Testing Conventions
## Imports
Always import from our harness:
```ts
import { render, screen, userEvent, waitFor } from 'tests/test-utils';
```
For API mocks:
```ts
import { server, rest } from 'mocks-server/server';
```
Do not import directly from `@testing-library/react`.
## Router
Use the router built into render:
```ts
render(<Page />, undefined, { initialRoute: '/traces-explorer' });
```
Only mock `useLocation` / `useParams` if the test depends on them.
## Hook Mocks
Pattern:
```ts
import useFoo from 'hooks/useFoo';
jest.mock('hooks/useFoo');
const mockUseFoo = jest.mocked(useFoo);
mockUseFoo.mockReturnValue(/* minimal shape */ as any);
```
Prefer helpers (`rqSuccess`, `rqLoading`, `rqError`) for React Query results.
## MSW
Global MSW server runs automatically.
Override per-test:
```ts
server.use(
rest.get('*/api/v1/foo', (_req, res, ctx) => res(ctx.status(200), ctx.json({ ok: true })))
);
```
Keep large responses in `mocks-server/__mockdata_`.
## Interactions
- Prefer `userEvent` for real user interactions (click, type, select, tab).
- Use `fireEvent` only for low-level/programmatic events not covered by `userEvent` (e.g., scroll, resize, setting `element.scrollTop` for virtualization). Wrap in `act(...)` if needed.
- Always await interactions:
```ts
const user = userEvent.setup({ pointerEventsCheck: 0 });
await user.click(screen.getByRole('button', { name: /save/i }));
```
```ts
// Example: virtualized list scroll (no userEvent helper)
const scroller = container.querySelector('[data-test-id="virtuoso-scroller"]') as HTMLElement;
scroller.scrollTop = targetScrollTop;
act(() => { fireEvent.scroll(scroller); });
```
## Timers
❌ No global fake timers.
✅ Per-test only, for debounce/throttle:
```ts
jest.useFakeTimers();
const user = userEvent.setup({ advanceTimers: (ms) => jest.advanceTimersByTime(ms) });
await user.type(screen.getByRole('textbox'), 'query');
jest.advanceTimersByTime(400);
jest.useRealTimers();
```
## Queries
Prefer accessible queries (`getByRole`, `findByRole`, `getByLabelText`).
Fallback: visible text.
Last resort: `data-testid`.
# Example Test (using only configured global mocks)
```ts
import { render, screen, userEvent, waitFor } from 'tests/test-utils';
import { server, rest } from 'mocks-server/server';
import MyComponent from '../MyComponent';
describe('MyComponent', () => {
it('renders and interacts', async () => {
const user = userEvent.setup({ pointerEventsCheck: 0 });
server.use(
rest.get('*/api/v1/example', (_req, res, ctx) => res(ctx.status(200), ctx.json({ value: 42 })))
);
render(<MyComponent />, undefined, { initialRoute: '/foo' });
expect(await screen.findByText(/value: 42/i)).toBeInTheDocument();
await user.click(screen.getByRole('button', { name: /refresh/i }));
await waitFor(() => expect(screen.getByText(/loading/i)).toBeInTheDocument());
});
});
```
# Anti-patterns
❌ Importing RTL directly
❌ Using global fake timers
❌ Wrapping render in `act(...)`
❌ Mocking infra dependencies locally (router, react-query)
✅ Use our harness (`tests/test-utils`)
✅ Use MSW for API overrides
✅ Use userEvent + await
✅ Pin time only in tests that assert relative dates
# Best Practices
- **Critical Functionality**: Prioritize testing business logic and utilities
- **Dependency Mocking**: Global mocks for infra, local mocks for business logic
- **Data Scenarios**: Always test valid, invalid, and edge cases
- **Descriptive Names**: Make test intent clear
- **Organization**: Group related tests in describe
- **Consistency**: Match repo conventions
- **Edge Cases**: Test null, undefined, unexpected values
- **Limit Scope**: 35 focused tests per file
- **Use Helpers**: `rqSuccess`, `makeUser`, etc.
- **No Any**: Enforce type safety
# Example Test
```ts
import { render, screen, userEvent, waitFor } from 'tests/test-utils';
import { server, rest } from 'mocks-server/server';
import MyComponent from '../MyComponent';
describe('MyComponent', () => {
it('renders and interacts', async () => {
const user = userEvent.setup({ pointerEventsCheck: 0 });
server.use(
rest.get('*/api/v1/example', (_req, res, ctx) => res(ctx.status(200), ctx.json({ value: 42 })))
);
render(<MyComponent />, undefined, { initialRoute: '/foo' });
expect(await screen.findByText(/value: 42/i)).toBeInTheDocument();
await user.click(screen.getByRole('button', { name: /refresh/i }));
await waitFor(() => expect(screen.getByText(/loading/i)).toBeInTheDocument());
});
});
```
# Anti-patterns
❌ Importing RTL directly
❌ Using global fake timers
❌ Wrapping render in `act(...)`
❌ Mocking infra dependencies locally (router, react-query)
✅ Use our harness (`tests/test-utils`)
✅ Use MSW for API overrides
✅ Use userEvent + await
✅ Pin time only in tests that assert relative dates
# TypeScript Type Safety Examples
## Proper Mock Typing
```ts
// ✅ GOOD - Properly typed mocks
interface User {
id: number;
name: string;
email: string;
}
interface ApiResponse<T> {
data: T;
status: number;
message: string;
}
// Type the mock functions
const mockFetchUser = jest.fn() as jest.MockedFunction<(id: number) => Promise<ApiResponse<User>>>;
const mockUpdateUser = jest.fn() as jest.MockedFunction<(user: User) => Promise<ApiResponse<User>>>;
// Mock implementation with proper typing
mockFetchUser.mockResolvedValue({
data: { id: 1, name: 'John Doe', email: 'john@example.com' },
status: 200,
message: 'Success'
});
// ❌ BAD - Using any type
const mockFetchUser = jest.fn() as any; // Don't do this
```
## React Component Testing with Types
```ts
// ✅ GOOD - Properly typed component testing
interface ComponentProps {
title: string;
data: User[];
onUserSelect: (user: User) => void;
isLoading?: boolean;
}
const TestComponent: React.FC<ComponentProps> = ({ title, data, onUserSelect, isLoading = false }) => {
// Component implementation
};
describe('TestComponent', () => {
it('should render with proper props', () => {
// Arrange - Type the props properly
const mockProps: ComponentProps = {
title: 'Test Title',
data: [{ id: 1, name: 'John', email: 'john@example.com' }],
onUserSelect: jest.fn() as jest.MockedFunction<(user: User) => void>,
isLoading: false
};
// Act
render(<TestComponent {...mockProps} />);
// Assert
expect(screen.getByText('Test Title')).toBeInTheDocument();
});
});
```
## Hook Testing with Types
```ts
// ✅ GOOD - Properly typed hook testing
interface UseUserDataReturn {
user: User | null;
loading: boolean;
error: string | null;
refetch: () => void;
}
const useUserData = (id: number): UseUserDataReturn => {
// Hook implementation
};
describe('useUserData', () => {
it('should return user data with proper typing', () => {
// Arrange
const mockUser: User = { id: 1, name: 'John', email: 'john@example.com' };
mockFetchUser.mockResolvedValue({
data: mockUser,
status: 200,
message: 'Success'
});
// Act
const { result } = renderHook(() => useUserData(1));
// Assert
expect(result.current.user).toEqual(mockUser);
expect(result.current.loading).toBe(false);
expect(result.current.error).toBeNull();
});
});
```
## Global Mock Type Safety
```ts
// ✅ GOOD - Type-safe global mocks
// In __mocks__/routerMock.ts
export const mockUseLocation = (overrides: Partial<Location> = {}): Location => ({
pathname: '/traces',
search: '',
hash: '',
state: null,
key: 'test-key',
...overrides,
});
// In test files
const location = useLocation(); // Properly typed from global mock
expect(location.pathname).toBe('/traces');
```
# TypeScript Configuration for Jest
## Required Jest Configuration
```json
// jest.config.ts
{
"preset": "ts-jest/presets/js-with-ts-esm",
"globals": {
"ts-jest": {
"useESM": true,
"isolatedModules": true,
"tsconfig": "<rootDir>/tsconfig.jest.json"
}
},
"extensionsToTreatAsEsm": [".ts", ".tsx"],
"moduleFileExtensions": ["ts", "tsx", "js", "json"]
}
```
## TypeScript Jest Configuration
```json
// tsconfig.jest.json
{
"extends": "./tsconfig.json",
"compilerOptions": {
"types": ["jest", "@testing-library/jest-dom"],
"esModuleInterop": true,
"allowSyntheticDefaultImports": true,
"moduleResolution": "node"
},
"include": [
"src/**/*",
"**/*.test.ts",
"**/*.test.tsx",
"__mocks__/**/*"
]
}
```
## Common Type Safety Patterns
### Mock Function Typing
```ts
// ✅ GOOD - Proper mock function typing
const mockApiCall = jest.fn() as jest.MockedFunction<typeof apiCall>;
const mockEventHandler = jest.fn() as jest.MockedFunction<(event: Event) => void>;
// ❌ BAD - Using any
const mockApiCall = jest.fn() as any;
```
### Generic Mock Typing
```ts
// ✅ GOOD - Generic mock typing
interface MockApiResponse<T> {
data: T;
status: number;
}
const mockFetchData = jest.fn() as jest.MockedFunction<
<T>(endpoint: string) => Promise<MockApiResponse<T>>
>;
// Usage
mockFetchData<User>('/users').mockResolvedValue({
data: { id: 1, name: 'John' },
status: 200
});
```
### React Testing Library with Types
```ts
// ✅ GOOD - Typed testing utilities
import { render, screen, RenderResult } from '@testing-library/react';
import { ComponentProps } from 'react';
type TestComponentProps = ComponentProps<typeof TestComponent>;
const renderTestComponent = (props: Partial<TestComponentProps> = {}): RenderResult => {
const defaultProps: TestComponentProps = {
title: 'Test',
data: [],
onSelect: jest.fn(),
...props
};
return render(<TestComponent {...defaultProps} />);
};
```
### Error Handling with Types
```ts
// ✅ GOOD - Typed error handling
interface ApiError {
message: string;
code: number;
details?: Record<string, unknown>;
}
const mockApiError: ApiError = {
message: 'API Error',
code: 500,
details: { endpoint: '/users' }
};
mockFetchUser.mockRejectedValue(new Error(JSON.stringify(mockApiError)));
```
## Type Safety Checklist
- [ ] All mock functions use `jest.MockedFunction<T>`
- [ ] All mock data has proper interfaces
- [ ] No `any` types in test files
- [ ] Generic types are used where appropriate
- [ ] Error types are properly defined
- [ ] Component props are typed
- [ ] Hook return types are defined
- [ ] API response types are defined
- [ ] Global mocks are type-safe
- [ ] Test utilities are properly typed
# Mock Decision Tree
```
Is it used in 20+ test files?
├─ YES → Use Global Mock
│ ├─ react-router-dom
│ ├─ react-query
│ ├─ antd components
│ └─ browser APIs
└─ NO → Is it business logic?
├─ YES → Use Local Mock
│ ├─ API endpoints
│ ├─ Custom hooks
│ └─ Domain components
└─ NO → Is it test-specific?
├─ YES → Use Local Mock
│ ├─ Error scenarios
│ ├─ Loading states
│ └─ Specific data
└─ NO → Consider Global Mock
└─ If it becomes frequently used
```
# Common Anti-Patterns to Avoid
❌ **Don't mock global dependencies locally:**
```js
// BAD - This is already globally mocked
jest.mock('react-router-dom', () => ({ ... }));
```
❌ **Don't create global mocks for test-specific data:**
```js
// BAD - This should be local
jest.mock('../api/tracesService', () => ({
getTraces: jest.fn(() => specificTestData)
}));
```
✅ **Do use global mocks for infrastructure:**
```js
// GOOD - Use global mock
import { useLocation } from 'react-router-dom';
```
✅ **Do create local mocks for business logic:**
```js
// GOOD - Local mock for specific test needs
jest.mock('../api/tracesService', () => ({
getTraces: jest.fn(() => mockTracesData)
}));
```

View File

@@ -18,7 +18,6 @@ import (
"github.com/SigNoz/signoz/pkg/modules/organization"
"github.com/SigNoz/signoz/pkg/modules/preference"
"github.com/SigNoz/signoz/pkg/modules/promote"
"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
"github.com/SigNoz/signoz/pkg/modules/session"
"github.com/SigNoz/signoz/pkg/modules/user"
"github.com/SigNoz/signoz/pkg/types"
@@ -45,7 +44,6 @@ type provider struct {
gatewayHandler gateway.Handler
fieldsHandler fields.Handler
authzHandler authz.Handler
rawDataExportHandler rawdataexport.Handler
}
func NewFactory(
@@ -65,7 +63,6 @@ func NewFactory(
gatewayHandler gateway.Handler,
fieldsHandler fields.Handler,
authzHandler authz.Handler,
rawDataExportHandler rawdataexport.Handler,
) factory.ProviderFactory[apiserver.APIServer, apiserver.Config] {
return factory.NewProviderFactory(factory.MustNewName("signoz"), func(ctx context.Context, providerSettings factory.ProviderSettings, config apiserver.Config) (apiserver.APIServer, error) {
return newProvider(
@@ -88,7 +85,6 @@ func NewFactory(
gatewayHandler,
fieldsHandler,
authzHandler,
rawDataExportHandler,
)
})
}
@@ -113,7 +109,6 @@ func newProvider(
gatewayHandler gateway.Handler,
fieldsHandler fields.Handler,
authzHandler authz.Handler,
rawDataExportHandler rawdataexport.Handler,
) (apiserver.APIServer, error) {
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/apiserver/signozapiserver")
router := mux.NewRouter().UseEncodedPath()
@@ -136,7 +131,6 @@ func newProvider(
gatewayHandler: gatewayHandler,
fieldsHandler: fieldsHandler,
authzHandler: authzHandler,
rawDataExportHandler: rawDataExportHandler,
}
provider.authZ = middleware.NewAuthZ(settings.Logger(), orgGetter, authz)
@@ -205,10 +199,6 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
return err
}
if err := provider.addRawDataExportRoutes(router); err != nil {
return err
}
return nil
}

View File

@@ -1,46 +0,0 @@
package signozapiserver
import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/types"
v5 "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/gorilla/mux"
)
func (provider *provider) addRawDataExportRoutes(router *mux.Router) error {
if err := router.Handle("/api/v1/export_raw_data", handler.New(provider.authZ.ViewAccess(provider.rawDataExportHandler.ExportRawData), handler.OpenAPIDef{
ID: "HandleExportRawDataGET",
Tags: []string{"logs", "traces"},
Summary: "Export raw data",
Description: "This endpoints allows simple query exporting raw data for traces and logs",
Request: new(types.ExportRawDataQueryParams),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest},
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/export_raw_data", handler.New(provider.authZ.ViewAccess(provider.rawDataExportHandler.ExportRawData), handler.OpenAPIDef{
ID: "HandleExportRawDataPOST",
Tags: []string{"logs", "traces"},
Summary: "Export raw data",
Description: "This endpoints allows complex query exporting raw data for traces and logs",
Request: new(struct {
v5.QueryRangeRequest
types.ExportRawDataFormatQueryParam
}),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest},
})).Methods(http.MethodPost).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -6,18 +6,17 @@ import (
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"sync"
"time"
"unicode"
"unicode/utf8"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
"github.com/SigNoz/signoz/pkg/types"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/types/authtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -35,27 +34,31 @@ func NewHandler(module rawdataexport.Module) rawdataexport.Handler {
// ExportRawData handles data export requests.
//
// API Documentation:
// Endpoint: GET /api/v1/export_raw_data - logs and traces only (simple queries via query params)
// Endpoint: POST /api/v1/export_raw_data - logs, traces, and trace operator (full QueryRangeRequest in JSON body)
// Endpoint: GET /api/v1/export_raw_data
//
// GET: Converts query params to QueryRangeRequest and delegates to common export logic. No composite_query/trace operator.
// Query Parameters:
//
// GET Query Parameters:
// - source (optional): Type of data to export ["logs" (default), "metrics", "traces"]
// Note: Currently only "logs" is fully supported
//
// - source (optional): ["logs" (default) or "traces"] - metrics not supported
// - format (optional): Output format ["csv" (default), "jsonl"]
// - start (required): Start time (Unix timestamp in nanoseconds)
// - end (required): End time (Unix timestamp in nanoseconds)
// - limit (optional): Max rows to export (cannot exceed MAX_EXPORT_ROW_COUNT_LIMIT)
// - filter (optional): Filter expression
// - columns (optional): Columns to include
// - order_by (optional): Sorting ["column:direction"]
//
// POST Request Body (QueryRangeRequest):
// - start (required): Start time for query (Unix timestamp in nanoseconds)
//
// - Accepts a full QueryRangeRequest JSON body with composite_query containing QueryEnvelope array
// - Supports builder_query and builder_trace_operator types
// - format query param still controls output format
// - end (required): End time for query (Unix timestamp in nanoseconds)
//
// - limit (optional): Maximum number of rows to export
// Constraints: Must be positive and cannot exceed MAX_EXPORT_ROW_COUNT_LIMIT
//
// - filter (optional): Filter expression to apply to the query
//
// - columns (optional): Specific columns to include in export
// Default: all columns are returned
// Format: ["context.field:type", "context.field", "field"]
//
// - order_by (optional): Sorting specification ["column:direction" or "context.field:type:direction"]
// Direction: "asc" or "desc"
// Default: ["timestamp:desc", "id:desc"]
//
// Response Headers:
// - Content-Type: "text/csv" or "application/x-ndjson"
@@ -83,45 +86,75 @@ func NewHandler(module rawdataexport.Module) rawdataexport.Handler {
// Export with filter and ordering:
// GET /api/v1/export_raw_data?start=1693612800000000000&end=1693699199000000000
// &filter=severity="error"&order_by=timestamp:desc&limit=1000
//
// Export with composite query (POST only):
// POST /api/v1/export_raw_data?format=csv
// Body: {"start":1693612800000000000,"end":1693699199000000000,"composite_query":{"queries":[...]}}
func (handler *handler) ExportRawData(rw http.ResponseWriter, r *http.Request) {
var queryRangeRequest qbtypes.QueryRangeRequest
var format string
if r.Method == http.MethodGet {
var params types.ExportRawDataQueryParams
if err := binding.Query.BindQuery(r.URL.Query(), &params); err != nil {
render.Error(rw, err)
return
}
format = params.Format
var err error
queryRangeRequest, err = buildQueryRangeRequest(&params)
if err != nil {
render.Error(rw, err)
return
}
} else {
var formatParam types.ExportRawDataFormatQueryParam
if err := binding.Query.BindQuery(r.URL.Query(), &formatParam); err != nil {
render.Error(rw, err)
return
}
format = formatParam.Format
if err := json.NewDecoder(r.Body).Decode(&queryRangeRequest); err != nil {
render.Error(rw, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid request body: %v", err))
return
}
}
if err := validateAndApplyExportLimits(&queryRangeRequest); err != nil {
source, err := getExportQuerySource(r.URL.Query())
if err != nil {
render.Error(rw, err)
return
}
switch source {
case "logs":
handler.exportLogs(rw, r)
case "traces":
handler.exportTraces(rw, r)
case "metrics":
handler.exportMetrics(rw, r)
default:
render.Error(rw, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid source: must be logs"))
}
}
func (handler *handler) exportMetrics(rw http.ResponseWriter, r *http.Request) {
render.Error(rw, errors.Newf(errors.TypeUnsupported, errors.CodeUnsupported, "metrics export is not yet supported"))
}
func (handler *handler) exportTraces(rw http.ResponseWriter, r *http.Request) {
render.Error(rw, errors.Newf(errors.TypeUnsupported, errors.CodeUnsupported, "traces export is not yet supported"))
}
func (handler *handler) exportLogs(rw http.ResponseWriter, r *http.Request) {
// Set up response headers
rw.Header().Set("Cache-Control", "no-cache")
rw.Header().Set("Vary", "Accept-Encoding") // Indicate that response varies based on Accept-Encoding
rw.Header().Set("Access-Control-Expose-Headers", "Content-Disposition, X-Response-Complete")
rw.Header().Set("Trailer", "X-Response-Complete")
rw.Header().Set("Transfer-Encoding", "chunked")
queryParams := r.URL.Query()
startTime, endTime, err := getExportQueryTimeRange(queryParams)
if err != nil {
render.Error(rw, err)
return
}
limit, err := getExportQueryLimit(queryParams)
if err != nil {
render.Error(rw, err)
return
}
format, err := getExportQueryFormat(queryParams)
if err != nil {
render.Error(rw, err)
return
}
// Set appropriate content type and filename
filename := fmt.Sprintf("data_exported_%s.%s", time.Now().Format("2006-01-02_150405"), format)
rw.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", filename))
filterExpression := queryParams.Get("filter")
orderByExpression, err := getExportQueryOrderBy(queryParams)
if err != nil {
render.Error(rw, err)
return
}
columns := getExportQueryColumns(queryParams)
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(rw, err)
@@ -134,145 +167,70 @@ func (handler *handler) ExportRawData(rw http.ResponseWriter, r *http.Request) {
return
}
setExportResponseHeaders(rw, format)
queryRangeRequest := qbtypes.QueryRangeRequest{
Start: startTime,
End: endTime,
RequestType: qbtypes.RequestTypeRaw,
CompositeQuery: qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: nil,
},
},
},
}
spec := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
Signal: telemetrytypes.SignalLogs,
Name: "raw",
Filter: &qbtypes.Filter{
Expression: filterExpression,
},
Limit: limit,
Order: orderByExpression,
}
spec.SelectFields = columns
queryRangeRequest.CompositeQuery.Queries[0].Spec = spec
// This will signal Export module to stop sending data
doneChan := make(chan any)
defer close(doneChan)
rowChan, errChan := handler.module.ExportRawData(r.Context(), orgID, &queryRangeRequest, doneChan)
isComplete, err := handler.executeExport(rowChan, errChan, format, rw)
if err != nil {
render.Error(rw, err)
return
}
rw.Header().Set("X-Response-Complete", strconv.FormatBool(isComplete))
}
var isComplete bool
// validateAndApplyExportLimits validates query types and applies default/max limits to all queries.
func validateAndApplyExportLimits(req *qbtypes.QueryRangeRequest) error {
isTraceOperatorQueryPresent := false
// Check if the trace operator query is present
queries := req.CompositeQuery.Queries
for idx := range len(queries) {
if _, ok := queries[idx].Spec.(qbtypes.QueryBuilderTraceOperator); ok {
isTraceOperatorQueryPresent = true
break
}
}
// If the trace operator query is not present, and there are multiple queries, return an error
if !isTraceOperatorQueryPresent && len(queries) > 1 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "multiple queries not allowed without a trace operator query")
}
for idx := range queries {
switch spec := queries[idx].Spec.(type) {
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
qbtypes.QueryBuilderTraceOperator:
limit := queries[idx].GetLimit()
if limit == 0 {
limit = DefaultExportRowCountLimit
} else if limit < 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be positive")
} else if limit > MaxExportRowCountLimit {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit cannot be more than %d", MaxExportRowCountLimit)
}
queries[idx].SetLimit(limit)
default:
return errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported query type: %T", spec)
}
}
return nil
}
// buildQueryRangeRequest builds a QueryRangeRequest from already-bound and validated GET query params.
func buildQueryRangeRequest(params *types.ExportRawDataQueryParams) (qbtypes.QueryRangeRequest, error) {
orderBy, err := parseExportQueryOrderBy(params.OrderBy)
if err != nil {
return qbtypes.QueryRangeRequest{}, err
}
columns := parseExportQueryColumns(params.Columns)
var filter *qbtypes.Filter
if params.Filter != "" {
filter = &qbtypes.Filter{Expression: params.Filter}
}
var query qbtypes.QueryEnvelope
switch params.Source {
case "logs":
query = qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
Signal: telemetrytypes.SignalLogs,
Filter: filter,
Limit: params.Limit,
Order: orderBy,
SelectFields: columns,
},
}
case "traces":
query = qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: filter,
Limit: params.Limit,
Order: orderBy,
SelectFields: columns,
},
}
}
return qbtypes.QueryRangeRequest{
Start: params.Start,
End: params.End,
RequestType: qbtypes.RequestTypeRaw,
CompositeQuery: qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{query},
},
}, nil
}
// setExportResponseHeaders sets common HTTP headers for export responses.
func setExportResponseHeaders(rw http.ResponseWriter, format string) {
rw.Header().Set("Cache-Control", "no-cache")
rw.Header().Set("Vary", "Accept-Encoding")
rw.Header().Set("Access-Control-Expose-Headers", "Content-Disposition, X-Response-Complete")
rw.Header().Set("Trailer", "X-Response-Complete")
rw.Header().Set("Transfer-Encoding", "chunked")
filename := fmt.Sprintf("data_exported_%s.%s", time.Now().Format("2006-01-02_150405"), format)
rw.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", filename))
}
// executeExport streams data from rowChan to the response writer in the specified format.
func (handler *handler) executeExport(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, format string, rw http.ResponseWriter) (bool, error) {
switch format {
case "csv", "":
rw.Header().Set("Content-Type", "text/csv")
csvWriter := csv.NewWriter(rw)
isComplete, err := handler.exportRawDataCSV(rowChan, errChan, csvWriter)
isComplete, err = handler.exportLogsCSV(rowChan, errChan, csvWriter)
if err != nil {
return false, err
render.Error(rw, err)
return
}
csvWriter.Flush()
return isComplete, nil
case "jsonl":
rw.Header().Set("Content-Type", "application/x-ndjson")
return handler.exportRawDataJSONL(rowChan, errChan, rw)
isComplete, err = handler.exportLogsJSONL(rowChan, errChan, rw)
if err != nil {
render.Error(rw, err)
return
}
default:
return false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid format: must be csv or jsonl")
render.Error(rw, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid format: must be csv or jsonl"))
return
}
rw.Header().Set("X-Response-Complete", strconv.FormatBool(isComplete))
}
// exportRawDataCSV is a generic CSV export function that works with any raw data (logs, traces, etc.)
func (handler *handler) exportRawDataCSV(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, csvWriter *csv.Writer) (bool, error) {
func (handler *handler) exportLogsCSV(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, csvWriter *csv.Writer) (bool, error) {
var header []string
headerToIndexMapping := make(map[string]int)
var once sync.Once
headerToIndexMapping := make(map[string]int, len(header))
totalBytes := uint64(0)
for {
@@ -281,14 +239,18 @@ func (handler *handler) exportRawDataCSV(rowChan <-chan *qbtypes.RawRow, errChan
if !ok {
return true, nil
}
once.Do(func() {
header := constructCSVHeaderFromQueryResponse(row.Data)
_ = csvWriter.Write(header)
// We ignore the error here, as it will get caught in the next iteration
if header == nil {
// Initialize and write header for CSV
header = constructCSVHeaderFromQueryResponse(row.Data)
if err := csvWriter.Write(header); err != nil {
return false, err
}
for i, col := range header {
headerToIndexMapping[col] = i
}
})
}
record := constructCSVRecordFromQueryResponse(row.Data, headerToIndexMapping)
if err := csvWriter.Write(record); err != nil {
return false, err
@@ -306,8 +268,8 @@ func (handler *handler) exportRawDataCSV(rowChan <-chan *qbtypes.RawRow, errChan
}
}
// exportRawDataJSONL is a generic JSONL export function that works with any raw data (logs, traces, etc.)
func (handler *handler) exportRawDataJSONL(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, writer io.Writer) (bool, error) {
func (handler *handler) exportLogsJSONL(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, writer io.Writer) (bool, error) {
totalBytes := uint64(0)
for {
select {
@@ -315,11 +277,9 @@ func (handler *handler) exportRawDataJSONL(rowChan <-chan *qbtypes.RawRow, errCh
if !ok {
return true, nil
}
jsonBytes, err := json.Marshal(row.Data)
if err != nil {
return false, errors.NewUnexpectedf(errors.CodeInternal, "error marshaling JSON: %s", err)
}
totalBytes += uint64(len(jsonBytes)) + 1
// Handle JSON format (JSONL - one object per line)
jsonBytes, _ := json.Marshal(row.Data)
totalBytes += uint64(len(jsonBytes)) + 1 // +1 for newline
if _, err := writer.Write(jsonBytes); err != nil {
return false, errors.NewUnexpectedf(errors.CodeInternal, "error writing JSON: %s", err)
@@ -339,6 +299,69 @@ func (handler *handler) exportRawDataJSONL(rowChan <-chan *qbtypes.RawRow, errCh
}
}
func getExportQuerySource(queryParams url.Values) (string, error) {
switch queryParams.Get("source") {
case "logs", "":
return "logs", nil
case "metrics":
return "metrics", errors.NewInvalidInputf(errors.CodeInvalidInput, "metrics export not yet supported")
case "traces":
return "traces", errors.NewInvalidInputf(errors.CodeInvalidInput, "traces export not yet supported")
default:
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid source: must be logs, metrics or traces")
}
}
func getExportQueryFormat(queryParams url.Values) (string, error) {
switch queryParams.Get("format") {
case "csv", "":
return "csv", nil
case "jsonl":
return "jsonl", nil
default:
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid format: must be csv or jsonl")
}
}
func getExportQueryLimit(queryParams url.Values) (int, error) {
limitStr := queryParams.Get("limit")
if limitStr == "" {
return DefaultExportRowCountLimit, nil
} else {
limit, err := strconv.Atoi(limitStr)
if err != nil {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid limit format: %s", err.Error())
}
if limit <= 0 {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be positive")
}
if limit > MaxExportRowCountLimit {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "limit cannot be more than %d", MaxExportRowCountLimit)
}
return limit, nil
}
}
func getExportQueryTimeRange(queryParams url.Values) (uint64, uint64, error) {
startTimeStr := queryParams.Get("start")
endTimeStr := queryParams.Get("end")
if startTimeStr == "" || endTimeStr == "" {
return 0, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "start and end time are required")
}
startTime, err := strconv.ParseUint(startTimeStr, 10, 64)
if err != nil {
return 0, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid start time format: %s", err.Error())
}
endTime, err := strconv.ParseUint(endTimeStr, 10, 64)
if err != nil {
return 0, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid end time format: %s", err.Error())
}
return startTime, endTime, nil
}
func constructCSVHeaderFromQueryResponse(data map[string]any) []string {
header := make([]string, 0, len(data))
for key := range data {
@@ -404,12 +427,9 @@ func constructCSVRecordFromQueryResponse(data map[string]any, headerToIndexMappi
valueStr = v.String()
default:
jsonBytes, err := json.Marshal(v)
if err != nil {
valueStr = fmt.Sprintf("%v", v)
} else {
valueStr = string(jsonBytes)
}
// For all other complex types (maps, structs, etc.)
jsonBytes, _ := json.Marshal(v)
valueStr = string(jsonBytes)
}
record[index] = sanitizeForCSV(valueStr)
@@ -418,17 +438,23 @@ func constructCSVRecordFromQueryResponse(data map[string]any, headerToIndexMappi
return record
}
// parseExportQueryColumns converts bound column strings to TelemetryFieldKey structs.
// Each column should be in the format "context.field:type" or "context.field" or "field"
func parseExportQueryColumns(columnParams []string) []telemetrytypes.TelemetryFieldKey {
// getExportQueryColumns parses the "columns" query parameters and returns a slice of TelemetryFieldKey structs.
// Each column should be a valid telemetry field key in the format "context.field:type" or "context.field" or "field"
func getExportQueryColumns(queryParams url.Values) []telemetrytypes.TelemetryFieldKey {
columnParams := queryParams["columns"]
columns := make([]telemetrytypes.TelemetryFieldKey, 0, len(columnParams))
for _, columnStr := range columnParams {
// Skip empty strings
columnStr = strings.TrimSpace(columnStr)
if columnStr == "" {
continue
}
columns = append(columns, telemetrytypes.GetFieldKeyFromKeyText(columnStr))
}
return columns
}
@@ -440,24 +466,51 @@ func getsizeOfStringSlice(slice []string) uint64 {
return totalBytes
}
// parseExportQueryOrderBy converts a bound order_by string to an OrderBy slice.
// The string should be in the format "column:direction" and is assumed already validated.
func parseExportQueryOrderBy(orderByParam string) ([]qbtypes.OrderBy, error) {
// getExportQueryOrderBy parses the "order_by" query parameters and returns a slice of OrderBy structs.
// Each "order_by" parameter should be in the format "column:direction"
// Each "column" should be a valid telemetry field key in the format "context.field:type" or "context.field" or "field"
func getExportQueryOrderBy(queryParams url.Values) ([]qbtypes.OrderBy, error) {
orderByParam := queryParams.Get("order_by")
orderByParam = strings.TrimSpace(orderByParam)
if orderByParam == "" {
return []qbtypes.OrderBy{}, nil
return telemetrylogs.DefaultLogsV2SortingOrder, nil
}
parts := strings.Split(orderByParam, ":")
if len(parts) != 2 && len(parts) != 3 {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order_by format: %s, should be <column>:<direction>", orderByParam)
}
column := strings.Join(parts[:len(parts)-1], ":")
direction := parts[len(parts)-1]
return []qbtypes.OrderBy{
orderDirection, ok := qbtypes.OrderDirectionMap[direction]
if !ok {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order_by direction: %s, should be one of %s, %s", direction, qbtypes.OrderDirectionAsc, qbtypes.OrderDirectionDesc)
}
orderByKey := telemetrytypes.GetFieldKeyFromKeyText(column)
orderBy := []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.GetFieldKeyFromKeyText(column),
TelemetryFieldKey: orderByKey,
},
Direction: qbtypes.OrderDirectionMap[direction],
Direction: orderDirection,
},
}, nil
}
// If we are ordering by the timestamp column, also order by the ID column
if orderByKey.Name == telemetrylogs.LogsV2TimestampColumn {
orderBy = append(orderBy, qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
Direction: orderDirection,
})
}
return orderBy, nil
}

View File

@@ -2,162 +2,261 @@ package implrawdataexport
import (
"net/url"
"strconv"
"testing"
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/types"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/stretchr/testify/assert"
)
func TestExportRawDataQueryParams_BindingDefaults(t *testing.T) {
var params types.ExportRawDataQueryParams
err := binding.Query.BindQuery(url.Values{}, &params)
assert.NoError(t, err)
assert.Equal(t, "logs", params.Source)
assert.Equal(t, "csv", params.Format)
assert.Equal(t, DefaultExportRowCountLimit, params.Limit)
}
func logQuery(limit int) qbtypes.QueryEnvelope {
return qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{Limit: limit},
}
}
func traceQuery(limit int) qbtypes.QueryEnvelope {
return qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{Limit: limit},
}
}
func traceOperatorQuery(limit int) qbtypes.QueryEnvelope {
return qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeTraceOperator,
Spec: qbtypes.QueryBuilderTraceOperator{Limit: limit},
}
}
func makeRequest(queries ...qbtypes.QueryEnvelope) qbtypes.QueryRangeRequest {
return qbtypes.QueryRangeRequest{
CompositeQuery: qbtypes.CompositeQuery{Queries: queries},
}
}
func TestValidateAndApplyExportLimits(t *testing.T) {
func TestGetExportQuerySource(t *testing.T) {
tests := []struct {
name string
req qbtypes.QueryRangeRequest
expectedError bool
// assertions on each query after the call (indexed)
checkQueries func(t *testing.T, queries []qbtypes.QueryEnvelope)
name string
queryParams url.Values
expectedSource string
expectedError bool
}{
{
name: "single log query, zero limit gets default",
req: makeRequest(logQuery(0)),
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, DefaultExportRowCountLimit, q[0].GetLimit())
},
name: "default logs source",
queryParams: url.Values{},
expectedSource: "logs",
expectedError: false,
},
{
name: "single log query, valid limit kept",
req: makeRequest(logQuery(1000)),
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, 1000, q[0].GetLimit())
},
name: "explicit logs source",
queryParams: url.Values{"source": {"logs"}},
expectedSource: "logs",
expectedError: false,
},
{
name: "single log query, max limit kept",
req: makeRequest(logQuery(MaxExportRowCountLimit)),
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, MaxExportRowCountLimit, q[0].GetLimit())
},
name: "metrics source - not supported",
queryParams: url.Values{"source": {"metrics"}},
expectedSource: "metrics",
expectedError: true,
},
{
name: "single log query, limit exceeds max",
req: makeRequest(logQuery(MaxExportRowCountLimit + 1)),
name: "traces source - not supported",
queryParams: url.Values{"source": {"traces"}},
expectedSource: "traces",
expectedError: true,
},
{
name: "invalid source",
queryParams: url.Values{"source": {"invalid"}},
expectedSource: "",
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
source, err := getExportQuerySource(tt.queryParams)
assert.Equal(t, tt.expectedSource, source)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
func TestGetExportQueryFormat(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedFormat string
expectedError bool
}{
{
name: "default csv format",
queryParams: url.Values{},
expectedFormat: "csv",
expectedError: false,
},
{
name: "explicit csv format",
queryParams: url.Values{"format": {"csv"}},
expectedFormat: "csv",
expectedError: false,
},
{
name: "jsonl format",
queryParams: url.Values{"format": {"jsonl"}},
expectedFormat: "jsonl",
expectedError: false,
},
{
name: "invalid format",
queryParams: url.Values{"format": {"xml"}},
expectedFormat: "",
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
format, err := getExportQueryFormat(tt.queryParams)
assert.Equal(t, tt.expectedFormat, format)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
func TestGetExportQueryLimit(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedLimit int
expectedError bool
}{
{
name: "default limit",
queryParams: url.Values{},
expectedLimit: DefaultExportRowCountLimit,
expectedError: false,
},
{
name: "valid limit",
queryParams: url.Values{"limit": {"5000"}},
expectedLimit: 5000,
expectedError: false,
},
{
name: "maximum limit",
queryParams: url.Values{"limit": {strconv.Itoa(MaxExportRowCountLimit)}},
expectedLimit: MaxExportRowCountLimit,
expectedError: false,
},
{
name: "limit exceeds maximum",
queryParams: url.Values{"limit": {"100000"}},
expectedLimit: 0,
expectedError: true,
},
{
name: "single log query, negative limit",
req: makeRequest(logQuery(-1)),
name: "invalid limit format",
queryParams: url.Values{"limit": {"invalid"}},
expectedLimit: 0,
expectedError: true,
},
{
name: "single trace query, zero limit gets default",
req: makeRequest(traceQuery(0)),
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, DefaultExportRowCountLimit, q[0].GetLimit())
},
},
{
name: "multiple queries without trace operator",
req: makeRequest(logQuery(0), traceQuery(0)),
expectedError: true,
},
{
name: "trace operator with other queries — non-operator disabled",
req: makeRequest(logQuery(500), traceOperatorQuery(1000)),
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.True(t, q[0].GetDisabled(), "log query should be disabled")
assert.False(t, q[1].GetDisabled(), "trace operator should stay enabled")
assert.Equal(t, 1000, q[1].GetLimit())
},
},
{
name: "trace operator alone, zero limit gets default",
req: makeRequest(traceOperatorQuery(0)),
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, DefaultExportRowCountLimit, q[0].GetLimit())
},
},
{
name: "unsupported query type",
req: makeRequest(qbtypes.QueryEnvelope{Type: qbtypes.QueryTypeBuilder, Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{}}),
name: "negative limit",
queryParams: url.Values{"limit": {"-100"}},
expectedLimit: 0,
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validateAndApplyExportLimits(&tt.req)
limit, err := getExportQueryLimit(tt.queryParams)
assert.Equal(t, tt.expectedLimit, limit)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
if tt.checkQueries != nil {
tt.checkQueries(t, tt.req.CompositeQuery.Queries)
}
}
})
}
}
func TestParseExportQueryColumns(t *testing.T) {
func TestGetExportQueryTimeRange(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedStartTime uint64
expectedEndTime uint64
expectedError bool
}{
{
name: "valid time range",
queryParams: url.Values{
"start": {"1640995200"},
"end": {"1641081600"},
},
expectedStartTime: 1640995200,
expectedEndTime: 1641081600,
expectedError: false,
},
{
name: "missing start time",
queryParams: url.Values{"end": {"1641081600"}},
expectedError: true,
},
{
name: "missing end time",
queryParams: url.Values{"start": {"1640995200"}},
expectedError: true,
},
{
name: "missing both times",
queryParams: url.Values{},
expectedError: true,
},
{
name: "invalid start time format",
queryParams: url.Values{
"start": {"invalid"},
"end": {"1641081600"},
},
expectedError: true,
},
{
name: "invalid end time format",
queryParams: url.Values{
"start": {"1640995200"},
"end": {"invalid"},
},
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
startTime, endTime, err := getExportQueryTimeRange(tt.queryParams)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
assert.Equal(t, tt.expectedStartTime, startTime)
assert.Equal(t, tt.expectedEndTime, endTime)
}
})
}
}
func TestGetExportQueryColumns(t *testing.T) {
tests := []struct {
name string
input []string
queryParams url.Values
expectedColumns []telemetrytypes.TelemetryFieldKey
}{
{
name: "empty input",
input: []string{},
name: "no columns specified",
queryParams: url.Values{},
expectedColumns: []telemetrytypes.TelemetryFieldKey{},
},
{
name: "single column",
input: []string{"timestamp"},
name: "single column",
queryParams: url.Values{
"columns": {"timestamp"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
},
},
{
name: "multiple columns",
input: []string{"timestamp", "message", "level"},
name: "multiple columns",
queryParams: url.Values{
"columns": {"timestamp", "message", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "message"},
@@ -165,90 +264,211 @@ func TestParseExportQueryColumns(t *testing.T) {
},
},
{
name: "empty entry is skipped",
input: []string{"timestamp", "", "level"},
name: "empty column name (should be skipped)",
queryParams: url.Values{
"columns": {"timestamp", "", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "level"},
},
},
{
name: "whitespace-only entry is skipped",
input: []string{"timestamp", " ", "level"},
name: "whitespace column name (should be skipped)",
queryParams: url.Values{
"columns": {"timestamp", " ", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "level"},
},
},
{
name: "column with context and type",
input: []string{"attribute.user:string"},
name: "valid column name with data type",
queryParams: url.Values{
"columns": {"timestamp", "attribute.user:string", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "user", FieldContext: telemetrytypes.FieldContextAttribute, FieldDataType: telemetrytypes.FieldDataTypeString},
{Name: "level"},
},
},
{
name: "column with context, dot-notation name",
input: []string{"attribute.user.string"},
name: "valid column name with dot notation",
queryParams: url.Values{
"columns": {"timestamp", "attribute.user.string", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "user.string", FieldContext: telemetrytypes.FieldContextAttribute},
{Name: "level"},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
columns := parseExportQueryColumns(tt.input)
columns := getExportQueryColumns(tt.queryParams)
assert.Equal(t, len(tt.expectedColumns), len(columns))
for i, expected := range tt.expectedColumns {
assert.Equal(t, expected, columns[i])
for i, expectedCol := range tt.expectedColumns {
assert.Equal(t, expectedCol, columns[i])
}
})
}
}
func TestParseExportQueryOrderBy(t *testing.T) {
func TestGetExportQueryOrderBy(t *testing.T) {
tests := []struct {
name string
input string
queryParams url.Values
expectedOrder []qbtypes.OrderBy
expectedError bool
}{
{
name: "empty string returns empty slice",
input: "",
expectedOrder: []qbtypes.OrderBy{},
expectedError: false,
},
{
name: "simple column asc",
input: "timestamp:asc",
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"},
},
},
},
expectedError: false,
},
{
name: "simple column desc",
input: "timestamp:desc",
name: "no order specified",
queryParams: url.Values{},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionDesc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "timestamp"},
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionDesc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "column with context and type qualifier",
input: "attribute.user:string:desc",
name: "single order error, direction not specified",
queryParams: url.Values{
"order_by": {"timestamp"},
},
expectedOrder: nil,
expectedError: true,
},
{
name: "single order no error",
queryParams: url.Values{
"order_by": {"timestamp:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "multiple orders",
queryParams: url.Values{
"order_by": {"timestamp:asc", "body:desc", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "empty order name (should be skipped)",
queryParams: url.Values{
"order_by": {"timestamp:asc", "", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "whitespace order name (should be skipped)",
queryParams: url.Values{
"order_by": {"timestamp:asc", " ", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "invalid order name (should error out)",
queryParams: url.Values{
"order_by": {"attributes.user:", "id:asc"},
},
expectedOrder: nil,
expectedError: true,
},
{
name: "valid order name (should be included)",
queryParams: url.Values{
"order_by": {"attribute.user:string:desc", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionDesc,
@@ -264,8 +484,10 @@ func TestParseExportQueryOrderBy(t *testing.T) {
expectedError: false,
},
{
name: "column with context, dot-notation name",
input: "attribute.user.string:desc",
name: "valid order name (should be included)",
queryParams: url.Values{
"order_by": {"attribute.user.string:desc", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionDesc,
@@ -279,35 +501,18 @@ func TestParseExportQueryOrderBy(t *testing.T) {
},
expectedError: false,
},
{
name: "resource with context and type",
input: "resource.service.name:string:asc",
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
},
},
expectedError: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
order, err := parseExportQueryOrderBy(tt.input)
order, err := getExportQueryOrderBy(tt.queryParams)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
assert.Equal(t, len(tt.expectedOrder), len(order))
for i, expected := range tt.expectedOrder {
assert.Equal(t, expected, order[i])
for i, expectedOrd := range tt.expectedOrder {
assert.Equal(t, expectedOrd, order[i])
}
}
})
@@ -324,8 +529,10 @@ func TestConstructCSVHeaderFromQueryResponse(t *testing.T) {
header := constructCSVHeaderFromQueryResponse(data)
// Since map iteration order is not guaranteed, check that all expected keys are present
expectedKeys := []string{"timestamp", "message", "level", "id"}
assert.Equal(t, len(expectedKeys), len(header))
for _, key := range expectedKeys {
assert.Contains(t, header, key)
}

View File

@@ -23,26 +23,8 @@ func NewModule(querier querier.Querier) rawdataexport.Module {
func (m *Module) ExportRawData(ctx context.Context, orgID valuer.UUID, rangeRequest *qbtypes.QueryRangeRequest, doneChan chan any) (chan *qbtypes.RawRow, chan error) {
isTraceOperatorQueryPresent := false
traceOperatorQueryIndex := -1
queries := rangeRequest.CompositeQuery.Queries
for idx := range len(queries) {
if _, ok := queries[idx].Spec.(qbtypes.QueryBuilderTraceOperator); ok {
isTraceOperatorQueryPresent = true
traceOperatorQueryIndex = idx
break
}
}
// If the trace operator query is present, mark the queries other than trace operator as disabled
if isTraceOperatorQueryPresent {
for idx := range len(queries) {
if idx != traceOperatorQueryIndex {
queries[idx].SetDisabled(true)
}
}
}
spec := rangeRequest.CompositeQuery.Queries[0].Spec.(qbtypes.QueryBuilderQuery[qbtypes.LogAggregation])
rowCountLimit := spec.Limit
rowChan := make(chan *qbtypes.RawRow, 1)
errChan := make(chan error, 1)
@@ -56,62 +38,52 @@ func (m *Module) ExportRawData(ctx context.Context, orgID valuer.UUID, rangeRequ
defer close(errChan)
defer close(rowChan)
if isTraceOperatorQueryPresent {
// If the trace operator query is present, we need to export the data for the trace operator query only
exportRawDataForSingleQuery(m.querier, contextWithTimeout, orgID, rangeRequest, rowChan, errChan, doneChan, traceOperatorQueryIndex)
} else {
// If the trace operator query is not present, we need to export the data for the first query only
exportRawDataForSingleQuery(m.querier, contextWithTimeout, orgID, rangeRequest, rowChan, errChan, doneChan, 0)
rowCount := 0
for rowCount < rowCountLimit {
spec.Limit = min(ChunkSize, rowCountLimit-rowCount)
spec.Offset = rowCount
rangeRequest.CompositeQuery.Queries[0].Spec = spec
response, err := m.querier.QueryRange(contextWithTimeout, orgID, rangeRequest)
if err != nil {
errChan <- err
return
}
newRowsCount := 0
for _, result := range response.Data.Results {
resultData, ok := result.(*qbtypes.RawData)
if !ok {
errChan <- errors.NewInternalf(errors.CodeInternal, "expected RawData, got %T", result)
return
}
newRowsCount += len(resultData.Rows)
for _, row := range resultData.Rows {
select {
case rowChan <- row:
case <-doneChan:
return
case <-ctx.Done():
errChan <- ctx.Err()
return
}
}
}
// Break if we did not receive any new rows
if newRowsCount == 0 {
return
}
rowCount += newRowsCount
}
}()
return rowChan, errChan
}
func exportRawDataForSingleQuery(querier querier.Querier, ctx context.Context, orgID valuer.UUID, rangeRequest *qbtypes.QueryRangeRequest, rowChan chan *qbtypes.RawRow, errChan chan error, doneChan chan any, queryIndex int) {
query := rangeRequest.CompositeQuery.Queries[queryIndex]
rowCountLimit := query.GetLimit()
rowCount := 0
for rowCount < rowCountLimit {
chunkSize := min(ChunkSize, rowCountLimit-rowCount)
query.SetLimit(chunkSize)
query.SetOffset(rowCount)
response, err := querier.QueryRange(ctx, orgID, rangeRequest)
if err != nil {
errChan <- err
return
}
newRowsCount := 0
for _, result := range response.Data.Results {
resultData, ok := result.(*qbtypes.RawData)
if !ok {
errChan <- errors.NewInternalf(errors.CodeInternal, "expected RawData, got %T", result)
return
}
newRowsCount += len(resultData.Rows)
for _, row := range resultData.Rows {
select {
case rowChan <- row:
case <-doneChan:
return
case <-ctx.Done():
errChan <- ctx.Err()
return
}
}
}
rowCount += newRowsCount
// Stop if we received fewer rows than requested — no more data available
if newRowsCount < chunkSize {
return
}
}
}

View File

@@ -586,7 +586,7 @@ func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
})).Methods(http.MethodGet)
// Export
// router.HandleFunc("/api/v1/export_raw_data", am.ViewAccess(aH.Signoz.Handlers.RawDataExport.ExportRawData)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/export_raw_data", am.ViewAccess(aH.Signoz.Handlers.RawDataExport.ExportRawData)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/span_percentile", am.ViewAccess(aH.Signoz.Handlers.SpanPercentile.GetSpanPercentileDetails)).Methods(http.MethodPost)

View File

@@ -871,7 +871,7 @@ func (v *filterExpressionVisitor) VisitKey(ctx *grammar.KeyContext) any {
// 1. either user meant key ( this is already handled above in fieldKeysForName )
// 2. or user meant `attribute.key` we look up in the map for all possible field keys with name 'attribute.key'
// Note:
// Note:
// If user only wants to search `attribute.key`, then they have to use `attribute.attribute.key`
// If user only wants to search `key`, then they have to use `key`
// If user wants to search both, they can use `attribute.key` and we will resolve the ambiguity

View File

@@ -20,7 +20,6 @@ import (
"github.com/SigNoz/signoz/pkg/modules/organization"
"github.com/SigNoz/signoz/pkg/modules/preference"
"github.com/SigNoz/signoz/pkg/modules/promote"
"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
"github.com/SigNoz/signoz/pkg/modules/session"
"github.com/SigNoz/signoz/pkg/modules/user"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
@@ -53,7 +52,6 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
struct{ gateway.Handler }{},
struct{ fields.Handler }{},
struct{ authz.Handler }{},
struct{ rawdataexport.Handler }{},
).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
if err != nil {
return nil, err

View File

@@ -252,7 +252,6 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
handlers.GatewayHandler,
handlers.Fields,
handlers.AuthzHandler,
handlers.RawDataExport,
),
)
}

View File

@@ -1,9 +1,6 @@
package telemetrytraces
import (
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
import "github.com/SigNoz/signoz/pkg/types/telemetrytypes"
var (
IntrinsicFields = map[string]telemetrytypes.TelemetryFieldKey{
@@ -382,19 +379,4 @@ var (
FieldDataType: telemetrytypes.FieldDataTypeString,
},
}
DefaultTracesSortingOrder = []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: DefaultFields["timestamp"],
},
Direction: qbtypes.OrderDirectionDesc,
},
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: DefaultFields["span_id"],
},
Direction: qbtypes.OrderDirectionDesc,
},
}
)

View File

@@ -2,8 +2,6 @@ package querybuildertypesv5
import (
"encoding/json"
"maps"
"reflect"
"strings"
"github.com/SigNoz/govaluate"
@@ -481,72 +479,3 @@ func (r *QueryRangeRequest) GetQueriesSupportingZeroDefault() map[string]bool {
return canDefaultZero
}
func (r *QueryRangeRequest) Copy() QueryRangeRequest {
c := QueryRangeRequest{
SchemaVersion: r.SchemaVersion,
Start: r.Start,
End: r.End,
RequestType: r.RequestType,
NoCache: r.NoCache,
FormatOptions: r.FormatOptions,
Variables: make(map[string]VariableItem),
CompositeQuery: CompositeQuery{},
}
maps.Copy(c.Variables, r.Variables)
c.CompositeQuery.Queries = make([]QueryEnvelope, len(r.CompositeQuery.Queries))
for i, q := range r.CompositeQuery.Queries {
if q.Spec != nil {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: spec.Copy(),
}
case QueryBuilderQuery[LogAggregation]:
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: spec.Copy(),
}
case QueryBuilderQuery[MetricAggregation]:
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: spec.Copy(),
}
case QueryBuilderFormula:
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: spec.Copy(),
}
case QueryBuilderJoin:
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: spec.Copy(),
}
case QueryBuilderTraceOperator:
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: spec.Copy(),
}
default:
// Check if spec implements a Copy method
specValue := reflect.ValueOf(q.Spec)
copyMethod := specValue.MethodByName("Copy")
if copyMethod.IsValid() && copyMethod.Type().NumIn() == 0 && copyMethod.Type().NumOut() == 1 {
// Call Copy method and use the result
result := copyMethod.Call(nil)
c.CompositeQuery.Queries[i] = QueryEnvelope{
Type: q.Type,
Spec: result[0].Interface(),
}
} else {
// Fallback to shallow copy
c.CompositeQuery.Queries[i] = q
}
}
}
}
return c
}

View File

@@ -1,386 +0,0 @@
package querybuildertypesv5
import "github.com/SigNoz/signoz/pkg/types/telemetrytypes"
// GetExpression returns the expression string. Panics for Join, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetExpression() string {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Expression
case QueryBuilderFormula:
return spec.Expression
}
panic("unsupported spec type")
}
// GetReturnSpansFrom returns the return-spans-from value. Panics for all types except TraceOperator.
func (q *QueryEnvelope) GetReturnSpansFrom() string {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.ReturnSpansFrom
}
panic("unsupported spec type")
}
// GetSignal returns the signal. Panics for all types except Query[T].
func (q *QueryEnvelope) GetSignal() telemetrytypes.Signal {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
return spec.Signal
case QueryBuilderQuery[LogAggregation]:
return spec.Signal
case QueryBuilderQuery[MetricAggregation]:
return spec.Signal
}
panic("unsupported spec type")
}
// GetSource returns the source. Panics for all types except Query[T].
func (q *QueryEnvelope) GetSource() telemetrytypes.Source {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
return spec.Source
case QueryBuilderQuery[LogAggregation]:
return spec.Source
case QueryBuilderQuery[MetricAggregation]:
return spec.Source
}
panic("unsupported spec type")
}
// GetQuery returns the raw query string. Panics for all types except PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetQuery() string {
switch spec := q.Spec.(type) {
case PromQuery:
return spec.Query
case ClickHouseQuery:
return spec.Query
}
panic("unsupported spec type")
}
// GetStep returns the PromQL step size. Panics for all types except PromQuery.
func (q *QueryEnvelope) GetStep() Step {
switch spec := q.Spec.(type) {
case PromQuery:
return spec.Step
}
panic("unsupported spec type")
}
// GetStats returns the PromQL stats flag. Panics for all types except PromQuery.
func (q *QueryEnvelope) GetStats() bool {
switch spec := q.Spec.(type) {
case PromQuery:
return spec.Stats
}
panic("unsupported spec type")
}
// GetLeft returns the left query reference of a join. Panics for all types except QueryBuilderJoin.
func (q *QueryEnvelope) GetLeft() QueryRef {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
return spec.Left
}
panic("unsupported spec type")
}
// GetRight returns the right query reference of a join. Panics for all types except QueryBuilderJoin.
func (q *QueryEnvelope) GetRight() QueryRef {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
return spec.Right
}
panic("unsupported spec type")
}
// GetJoinType returns the join type. Panics for all types except QueryBuilderJoin.
func (q *QueryEnvelope) GetJoinType() JoinType {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
return spec.Type
}
panic("unsupported spec type")
}
// GetOn returns the join ON condition. Panics for all types except QueryBuilderJoin.
func (q *QueryEnvelope) GetOn() string {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
return spec.On
}
panic("unsupported spec type")
}
// GetQueryName returns the name of the spec.
func (q *QueryEnvelope) GetQueryName() string {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Name
case QueryBuilderQuery[TraceAggregation]:
return spec.Name
case QueryBuilderQuery[LogAggregation]:
return spec.Name
case QueryBuilderQuery[MetricAggregation]:
return spec.Name
case QueryBuilderFormula:
return spec.Name
case QueryBuilderJoin:
return spec.Name
case PromQuery:
return spec.Name
case ClickHouseQuery:
return spec.Name
}
panic("unsupported spec type")
}
// GetDisabled returns whether the spec is disabled.
func (q *QueryEnvelope) GetDisabled() bool {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Disabled
case QueryBuilderQuery[TraceAggregation]:
return spec.Disabled
case QueryBuilderQuery[LogAggregation]:
return spec.Disabled
case QueryBuilderQuery[MetricAggregation]:
return spec.Disabled
case QueryBuilderFormula:
return spec.Disabled
case QueryBuilderJoin:
return spec.Disabled
case PromQuery:
return spec.Disabled
case ClickHouseQuery:
return spec.Disabled
}
panic("unsupported spec type")
}
// GetLimit returns the row limit. Panics for PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetLimit() int {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Limit
case QueryBuilderQuery[TraceAggregation]:
return spec.Limit
case QueryBuilderQuery[LogAggregation]:
return spec.Limit
case QueryBuilderQuery[MetricAggregation]:
return spec.Limit
case QueryBuilderFormula:
return spec.Limit
case QueryBuilderJoin:
return spec.Limit
}
panic("unsupported spec type")
}
// GetOffset returns the row offset. Panics for Formula, Join, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetOffset() int {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Offset
case QueryBuilderQuery[TraceAggregation]:
return spec.Offset
case QueryBuilderQuery[LogAggregation]:
return spec.Offset
case QueryBuilderQuery[MetricAggregation]:
return spec.Offset
}
panic("unsupported spec type")
}
// GetType returns the QueryType of the envelope.
func (q *QueryEnvelope) GetType() QueryType {
return q.Type
}
// GetOrder returns the order-by clauses. Panics for PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetOrder() []OrderBy {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Order
case QueryBuilderQuery[TraceAggregation]:
return spec.Order
case QueryBuilderQuery[LogAggregation]:
return spec.Order
case QueryBuilderQuery[MetricAggregation]:
return spec.Order
case QueryBuilderFormula:
return spec.Order
case QueryBuilderJoin:
return spec.Order
}
panic("unsupported spec type")
}
// GetGroupBy returns the group-by keys. Panics for Formula, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetGroupBy() []GroupByKey {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.GroupBy
case QueryBuilderQuery[TraceAggregation]:
return spec.GroupBy
case QueryBuilderQuery[LogAggregation]:
return spec.GroupBy
case QueryBuilderQuery[MetricAggregation]:
return spec.GroupBy
case QueryBuilderJoin:
return spec.GroupBy
}
panic("unsupported spec type")
}
// GetFilter returns the filter. Panics for Formula, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetFilter() *Filter {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Filter
case QueryBuilderQuery[TraceAggregation]:
return spec.Filter
case QueryBuilderQuery[LogAggregation]:
return spec.Filter
case QueryBuilderQuery[MetricAggregation]:
return spec.Filter
case QueryBuilderJoin:
return spec.Filter
}
panic("unsupported spec type")
}
// GetHaving returns the having clause. Panics for PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetHaving() *Having {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Having
case QueryBuilderQuery[TraceAggregation]:
return spec.Having
case QueryBuilderQuery[LogAggregation]:
return spec.Having
case QueryBuilderQuery[MetricAggregation]:
return spec.Having
case QueryBuilderFormula:
return spec.Having
case QueryBuilderJoin:
return spec.Having
}
panic("unsupported spec type")
}
// GetFunctions returns the post-processing functions. Panics for PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetFunctions() []Function {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Functions
case QueryBuilderQuery[TraceAggregation]:
return spec.Functions
case QueryBuilderQuery[LogAggregation]:
return spec.Functions
case QueryBuilderQuery[MetricAggregation]:
return spec.Functions
case QueryBuilderFormula:
return spec.Functions
case QueryBuilderJoin:
return spec.Functions
}
panic("unsupported spec type")
}
// GetSelectFields returns the selected fields. Panics for Formula, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetSelectFields() []telemetrytypes.TelemetryFieldKey {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.SelectFields
case QueryBuilderQuery[TraceAggregation]:
return spec.SelectFields
case QueryBuilderQuery[LogAggregation]:
return spec.SelectFields
case QueryBuilderQuery[MetricAggregation]:
return spec.SelectFields
case QueryBuilderJoin:
return spec.SelectFields
}
panic("unsupported spec type")
}
// GetLegend returns the legend label. Panics for QueryBuilderJoin.
func (q *QueryEnvelope) GetLegend() string {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Legend
case QueryBuilderQuery[TraceAggregation]:
return spec.Legend
case QueryBuilderQuery[LogAggregation]:
return spec.Legend
case QueryBuilderQuery[MetricAggregation]:
return spec.Legend
case QueryBuilderFormula:
return spec.Legend
case PromQuery:
return spec.Legend
case ClickHouseQuery:
return spec.Legend
}
panic("unsupported spec type")
}
// GetCursor returns the pagination cursor. Panics for Formula, Join, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetCursor() string {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.Cursor
case QueryBuilderQuery[TraceAggregation]:
return spec.Cursor
case QueryBuilderQuery[LogAggregation]:
return spec.Cursor
case QueryBuilderQuery[MetricAggregation]:
return spec.Cursor
}
panic("unsupported spec type")
}
// GetStepInterval returns the step interval. Panics for Formula, Join, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetStepInterval() Step {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
return spec.StepInterval
case QueryBuilderQuery[TraceAggregation]:
return spec.StepInterval
case QueryBuilderQuery[LogAggregation]:
return spec.StepInterval
case QueryBuilderQuery[MetricAggregation]:
return spec.StepInterval
}
panic("unsupported spec type")
}
// GetSecondaryAggregations returns the secondary aggregations. Panics for TraceOperator, Formula, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetSecondaryAggregations() []SecondaryAggregation {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
return spec.SecondaryAggregations
case QueryBuilderQuery[LogAggregation]:
return spec.SecondaryAggregations
case QueryBuilderQuery[MetricAggregation]:
return spec.SecondaryAggregations
case QueryBuilderJoin:
return spec.SecondaryAggregations
}
panic("unsupported spec type")
}
// GetLimitBy returns the limit-by configuration. Panics for TraceOperator, Formula, Join, PromQuery and ClickHouseQuery.
func (q *QueryEnvelope) GetLimitBy() *LimitBy {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
return spec.LimitBy
case QueryBuilderQuery[LogAggregation]:
return spec.LimitBy
case QueryBuilderQuery[MetricAggregation]:
return spec.LimitBy
}
panic("unsupported spec type")
}

View File

@@ -1,458 +0,0 @@
package querybuildertypesv5
import "github.com/SigNoz/signoz/pkg/types/telemetrytypes"
// SetExpression sets the expression string of the spec, if applicable.
func (q *QueryEnvelope) SetExpression(expression string) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Expression = expression
q.Spec = spec
case QueryBuilderFormula:
spec.Expression = expression
q.Spec = spec
}
}
// SetReturnSpansFrom sets the return-spans-from value, if applicable.
func (q *QueryEnvelope) SetReturnSpansFrom(returnSpansFrom string) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.ReturnSpansFrom = returnSpansFrom
q.Spec = spec
}
}
// SetSignal sets the signal of the spec, if applicable.
func (q *QueryEnvelope) SetSignal(signal telemetrytypes.Signal) {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
spec.Signal = signal
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Signal = signal
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Signal = signal
q.Spec = spec
}
}
// SetSource sets the source of the spec, if applicable.
func (q *QueryEnvelope) SetSource(source telemetrytypes.Source) {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
spec.Source = source
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Source = source
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Source = source
q.Spec = spec
}
}
// SetQuery sets the raw query string of the spec, if applicable.
func (q *QueryEnvelope) SetQuery(query string) {
switch spec := q.Spec.(type) {
case PromQuery:
spec.Query = query
q.Spec = spec
case ClickHouseQuery:
spec.Query = query
q.Spec = spec
}
}
// SetStep sets the PromQL step size, if applicable.
func (q *QueryEnvelope) SetStep(step Step) {
switch spec := q.Spec.(type) {
case PromQuery:
spec.Step = step
q.Spec = spec
}
}
// SetStats sets the PromQL stats flag, if applicable.
func (q *QueryEnvelope) SetStats(stats bool) {
switch spec := q.Spec.(type) {
case PromQuery:
spec.Stats = stats
q.Spec = spec
}
}
// SetLeft sets the left query reference of a join, if applicable.
func (q *QueryEnvelope) SetLeft(left QueryRef) {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
spec.Left = left
q.Spec = spec
}
}
// SetRight sets the right query reference of a join, if applicable.
func (q *QueryEnvelope) SetRight(right QueryRef) {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
spec.Right = right
q.Spec = spec
}
}
// SetJoinType sets the join type, if applicable.
func (q *QueryEnvelope) SetJoinType(joinType JoinType) {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
spec.Type = joinType
q.Spec = spec
}
}
// SetOn sets the join ON condition, if applicable.
func (q *QueryEnvelope) SetOn(on string) {
switch spec := q.Spec.(type) {
case QueryBuilderJoin:
spec.On = on
q.Spec = spec
}
}
// SetQueryName sets the name of the spec, if applicable.
func (q *QueryEnvelope) SetQueryName(name string) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Name = name
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Name = name
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Name = name
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Name = name
q.Spec = spec
case QueryBuilderFormula:
spec.Name = name
q.Spec = spec
case QueryBuilderJoin:
spec.Name = name
q.Spec = spec
case PromQuery:
spec.Name = name
q.Spec = spec
case ClickHouseQuery:
spec.Name = name
q.Spec = spec
}
}
// SetDisabled sets the disabled flag of the spec, if applicable.
func (q *QueryEnvelope) SetDisabled(disabled bool) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Disabled = disabled
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Disabled = disabled
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Disabled = disabled
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Disabled = disabled
q.Spec = spec
case QueryBuilderFormula:
spec.Disabled = disabled
q.Spec = spec
case QueryBuilderJoin:
spec.Disabled = disabled
q.Spec = spec
case PromQuery:
spec.Disabled = disabled
q.Spec = spec
case ClickHouseQuery:
spec.Disabled = disabled
q.Spec = spec
}
}
// SetLimit sets the row limit of the spec, if applicable.
func (q *QueryEnvelope) SetLimit(limit int) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Limit = limit
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Limit = limit
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Limit = limit
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Limit = limit
q.Spec = spec
case QueryBuilderFormula:
spec.Limit = limit
q.Spec = spec
case QueryBuilderJoin:
spec.Limit = limit
q.Spec = spec
}
}
// SetOffset sets the row offset of the spec, if applicable.
func (q *QueryEnvelope) SetOffset(offset int) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Offset = offset
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Offset = offset
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Offset = offset
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Offset = offset
q.Spec = spec
}
}
// SetType sets the QueryType of the envelope.
func (q *QueryEnvelope) SetType(t QueryType) {
q.Type = t
}
// SetOrder sets the order-by clauses of the spec, if applicable.
func (q *QueryEnvelope) SetOrder(order []OrderBy) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Order = order
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Order = order
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Order = order
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Order = order
q.Spec = spec
case QueryBuilderFormula:
spec.Order = order
q.Spec = spec
case QueryBuilderJoin:
spec.Order = order
q.Spec = spec
}
}
// SetGroupBy sets the group-by keys of the spec, if applicable.
func (q *QueryEnvelope) SetGroupBy(groupBy []GroupByKey) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.GroupBy = groupBy
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.GroupBy = groupBy
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.GroupBy = groupBy
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.GroupBy = groupBy
q.Spec = spec
case QueryBuilderJoin:
spec.GroupBy = groupBy
q.Spec = spec
}
}
// SetFilter sets the filter of the spec, if applicable.
func (q *QueryEnvelope) SetFilter(filter *Filter) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Filter = filter
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Filter = filter
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Filter = filter
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Filter = filter
q.Spec = spec
case QueryBuilderJoin:
spec.Filter = filter
q.Spec = spec
}
}
// SetHaving sets the having clause of the spec, if applicable.
func (q *QueryEnvelope) SetHaving(having *Having) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Having = having
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Having = having
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Having = having
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Having = having
q.Spec = spec
case QueryBuilderFormula:
spec.Having = having
q.Spec = spec
case QueryBuilderJoin:
spec.Having = having
q.Spec = spec
}
}
// SetFunctions sets the post-processing functions of the spec, if applicable.
func (q *QueryEnvelope) SetFunctions(functions []Function) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Functions = functions
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Functions = functions
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Functions = functions
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Functions = functions
q.Spec = spec
case QueryBuilderFormula:
spec.Functions = functions
q.Spec = spec
case QueryBuilderJoin:
spec.Functions = functions
q.Spec = spec
}
}
// SetSelectFields sets the selected fields of the spec, if applicable.
func (q *QueryEnvelope) SetSelectFields(fields []telemetrytypes.TelemetryFieldKey) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.SelectFields = fields
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.SelectFields = fields
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.SelectFields = fields
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.SelectFields = fields
q.Spec = spec
case QueryBuilderJoin:
spec.SelectFields = fields
q.Spec = spec
}
}
// SetLegend sets the legend label of the spec, if applicable.
func (q *QueryEnvelope) SetLegend(legend string) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Legend = legend
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Legend = legend
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Legend = legend
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Legend = legend
q.Spec = spec
case QueryBuilderFormula:
spec.Legend = legend
q.Spec = spec
case PromQuery:
spec.Legend = legend
q.Spec = spec
case ClickHouseQuery:
spec.Legend = legend
q.Spec = spec
}
}
// SetCursor sets the pagination cursor of the spec, if applicable.
func (q *QueryEnvelope) SetCursor(cursor string) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.Cursor = cursor
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.Cursor = cursor
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.Cursor = cursor
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.Cursor = cursor
q.Spec = spec
}
}
// SetStepInterval sets the step interval of the spec, if applicable.
func (q *QueryEnvelope) SetStepInterval(step Step) {
switch spec := q.Spec.(type) {
case QueryBuilderTraceOperator:
spec.StepInterval = step
q.Spec = spec
case QueryBuilderQuery[TraceAggregation]:
spec.StepInterval = step
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.StepInterval = step
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.StepInterval = step
q.Spec = spec
}
}
// SetSecondaryAggregations sets the secondary aggregations of the spec, if applicable.
func (q *QueryEnvelope) SetSecondaryAggregations(secondaryAggregations []SecondaryAggregation) {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
spec.SecondaryAggregations = secondaryAggregations
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.SecondaryAggregations = secondaryAggregations
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.SecondaryAggregations = secondaryAggregations
q.Spec = spec
case QueryBuilderJoin:
spec.SecondaryAggregations = secondaryAggregations
q.Spec = spec
}
}
// SetLimitBy sets the limit-by configuration of the spec, if applicable.
func (q *QueryEnvelope) SetLimitBy(limitBy *LimitBy) {
switch spec := q.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
spec.LimitBy = limitBy
q.Spec = spec
case QueryBuilderQuery[LogAggregation]:
spec.LimitBy = limitBy
q.Spec = spec
case QueryBuilderQuery[MetricAggregation]:
spec.LimitBy = limitBy
q.Spec = spec
}
}

View File

@@ -10,9 +10,55 @@ import (
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// queryName returns the name from any query envelope spec type.
func (e QueryEnvelope) queryName() string {
switch spec := e.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
return spec.Name
case QueryBuilderQuery[LogAggregation]:
return spec.Name
case QueryBuilderQuery[MetricAggregation]:
return spec.Name
case QueryBuilderFormula:
return spec.Name
case QueryBuilderTraceOperator:
return spec.Name
case QueryBuilderJoin:
return spec.Name
case PromQuery:
return spec.Name
case ClickHouseQuery:
return spec.Name
}
return ""
}
// isDisabled returns the disabled status from any query envelope spec type.
func (e QueryEnvelope) isDisabled() bool {
switch spec := e.Spec.(type) {
case QueryBuilderQuery[TraceAggregation]:
return spec.Disabled
case QueryBuilderQuery[LogAggregation]:
return spec.Disabled
case QueryBuilderQuery[MetricAggregation]:
return spec.Disabled
case QueryBuilderFormula:
return spec.Disabled
case QueryBuilderTraceOperator:
return spec.Disabled
case QueryBuilderJoin:
return spec.Disabled
case PromQuery:
return spec.Disabled
case ClickHouseQuery:
return spec.Disabled
}
return false
}
// getQueryIdentifier returns a friendly identifier for a query based on its type and name/content
func getQueryIdentifier(envelope QueryEnvelope, index int) string {
name := envelope.GetQueryName()
name := envelope.queryName()
var typeLabel string
switch envelope.Type {
@@ -425,7 +471,7 @@ func (r *QueryRangeRequest) Validate() error {
// validateAllQueriesNotDisabled validates that at least one query in the composite query is enabled
func (r *QueryRangeRequest) validateAllQueriesNotDisabled() error {
for _, envelope := range r.CompositeQuery.Queries {
if !envelope.GetDisabled() {
if !envelope.isDisabled() {
return nil
}
}
@@ -460,7 +506,7 @@ func (c *CompositeQuery) Validate(requestType RequestType) error {
// Check name uniqueness for builder queries
if envelope.Type == QueryTypeBuilder || envelope.Type == QueryTypeSubQuery {
name := envelope.GetQueryName()
name := envelope.queryName()
if name != "" {
if queryNames[name] {
return errors.NewInvalidInputf(

View File

@@ -816,7 +816,7 @@ func TestQueryEnvelope_Helpers(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := tt.envelope.GetQueryName()
got := tt.envelope.queryName()
if got != tt.want {
t.Errorf("queryName() = %q, want %q", got, tt.want)
}
@@ -868,7 +868,7 @@ func TestQueryEnvelope_Helpers(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := tt.envelope.GetDisabled()
got := tt.envelope.isDisabled()
if got != tt.want {
t.Errorf("isDisabled() = %v, want %v", got, tt.want)
}

View File

@@ -1,35 +0,0 @@
package types
// ExportRawDataQueryParams represents the query parameters for the export raw data endpoint
type ExportRawDataQueryParams struct {
ExportRawDataFormatQueryParam
// Source specifies the type of data to export: "logs" or "traces"
Source string `query:"source,default=logs" default:"logs" enum:"logs,traces"`
// Start is the start time for the query (Unix timestamp in nanoseconds)
Start uint64 `query:"start"`
// End is the end time for the query (Unix timestamp in nanoseconds)
End uint64 `query:"end"`
// Limit specifies the maximum number of rows to export
Limit int `query:"limit,default=10000" default:"10000" minimum:"1" maximum:"50000"`
// Filter is a filter expression to apply to the query
Filter string `query:"filter"`
// Columns specifies the columns to include in the export
// Format: ["context.field:type", "context.field", "field"]
Columns []string `query:"columns"`
// OrderBy specifies the sorting order
// Format: "column:direction" or "context.field:type:direction"
// Direction can be "asc" or "desc"
OrderBy string `query:"order_by"`
}
type ExportRawDataFormatQueryParam struct {
// Format specifies the output format: "csv" or "jsonl"
Format string `query:"format,default=csv" default:"csv" enum:"csv,jsonl"`
}

View File

@@ -0,0 +1,62 @@
from http import HTTPStatus
from typing import Callable
import requests
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.types import SigNoz
DUPLICATE_USER_EMAIL = "duplicate@integration.test"
def test_duplicate_user_invite_rejected(
signoz: SigNoz,
get_token: Callable[[str, str], str],
):
"""
Verify that the unique index on (email, org_id) in the users table prevents
creating duplicate users. This invites a new user, accepts the invite, then
tries to invite and accept the same email again expecting a failure.
"""
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Step 1: Invite a new user.
initial_invite_response = requests.post(
signoz.self.host_configs["8080"].get("/api/v1/invite"),
json={"email": DUPLICATE_USER_EMAIL, "role": "EDITOR"},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=2,
)
assert initial_invite_response.status_code == HTTPStatus.CREATED
initial_invite_token = initial_invite_response.json()["data"]["token"]
# Step 2: Accept the invite to create the user.
initial_accept_response = requests.post(
signoz.self.host_configs["8080"].get("/api/v1/invite/accept"),
json={"token": initial_invite_token, "password": "password123Z$"},
timeout=2,
)
assert initial_accept_response.status_code == HTTPStatus.CREATED
# Step 3: Invite the same email again.
duplicate_invite_response = requests.post(
signoz.self.host_configs["8080"].get("/api/v1/invite"),
json={"email": DUPLICATE_USER_EMAIL, "role": "VIEWER"},
headers={"Authorization": f"Bearer {admin_token}"},
timeout=2,
)
# The invite creation itself may be rejected if the app checks for existing users.
if duplicate_invite_response.status_code != HTTPStatus.CREATED:
assert duplicate_invite_response.status_code == HTTPStatus.CONFLICT
return
duplicate_invite_token = duplicate_invite_response.json()["data"]["token"]
# Step 4: Accept the duplicate invite — should fail due to unique constraint.
duplicate_accept_response = requests.post(
signoz.self.host_configs["8080"].get("/api/v1/invite/accept"),
json={"token": duplicate_invite_token, "password": "password123Z$"},
timeout=2,
)
assert duplicate_accept_response.status_code == HTTPStatus.CONFLICT

View File

@@ -1,617 +0,0 @@
from datetime import datetime, timedelta, timezone
from http import HTTPStatus
from typing import Callable, List
from urllib.parse import urlencode
import csv
import io
import json
import requests
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.logs import Logs
def test_export_logs_csv(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert 3 logs with different severity levels and attributes.
Tests:
1. Export logs as CSV format
2. Verify CSV structure and content
3. Validate headers are present
4. Check log data is correctly formatted
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=10),
body="Application started successfully",
severity_text="INFO",
resources={
"service.name": "api-service",
"deployment.environment": "production",
"host.name": "server-01",
},
attributes={
"http.method": "GET",
"http.status_code": 200,
"user.id": "user123",
},
),
Logs(
timestamp=now - timedelta(seconds=8),
body="Connection to database failed",
severity_text="ERROR",
resources={
"service.name": "api-service",
"deployment.environment": "production",
"host.name": "server-01",
},
attributes={
"error.type": "ConnectionError",
"db.name": "production_db",
},
),
Logs(
timestamp=now - timedelta(seconds=5),
body="Request processed",
severity_text="DEBUG",
resources={
"service.name": "worker-service",
"deployment.environment": "production",
"host.name": "server-02",
},
attributes={
"request.id": "req-456",
"duration_ms": 150.5,
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
}
# Export logs as CSV (default format, no source needed)
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "text/csv"
assert "attachment" in response.headers.get("Content-Disposition", "")
# Parse CSV content
csv_content = response.text
csv_reader = csv.DictReader(io.StringIO(csv_content))
rows = list(csv_reader)
assert len(rows) == 3, f"Expected 3 rows, got {len(rows)}"
# Verify log bodies are present in the exported data
bodies = [row.get("body") for row in rows]
assert "Application started successfully" in bodies
assert "Connection to database failed" in bodies
assert "Request processed" in bodies
# Verify severity levels
severities = [row.get("severity_text") for row in rows]
assert "INFO" in severities
assert "ERROR" in severities
assert "DEBUG" in severities
def test_export_logs_jsonl(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert 2 logs with different attributes.
Tests:
1. Export logs as JSONL format
2. Verify JSONL structure and content
3. Check each line is valid JSON
4. Validate log data is correctly formatted
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=10),
body="User logged in",
severity_text="INFO",
resources={
"service.name": "auth-service",
"deployment.environment": "staging",
},
attributes={
"user.email": "test@example.com",
"session.id": "sess-789",
},
),
Logs(
timestamp=now - timedelta(seconds=5),
body="Payment processed successfully",
severity_text="INFO",
resources={
"service.name": "payment-service",
"deployment.environment": "staging",
},
attributes={
"transaction.id": "txn-123",
"amount": 99.99,
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "jsonl",
"source": "logs",
}
# Export logs as JSONL
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
assert "attachment" in response.headers.get("Content-Disposition", "")
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 2, f"Expected 2 lines, got {len(jsonl_lines)}"
# Verify each line is valid JSON
json_objects = []
for line in jsonl_lines:
obj = json.loads(line)
json_objects.append(obj)
assert "id" in obj
assert "timestamp" in obj
assert "body" in obj
assert "severity_text" in obj
# Verify log bodies
bodies = [obj.get("body") for obj in json_objects]
assert "User logged in" in bodies
assert "Payment processed successfully" in bodies
def test_export_logs_with_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert logs with different severity levels.
Tests:
1. Export logs with filter applied
2. Verify only filtered logs are returned
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=10),
body="Info message",
severity_text="INFO",
resources={
"service.name": "test-service",
},
attributes={},
),
Logs(
timestamp=now - timedelta(seconds=8),
body="Error message",
severity_text="ERROR",
resources={
"service.name": "test-service",
},
attributes={},
),
Logs(
timestamp=now - timedelta(seconds=5),
body="Another error message",
severity_text="ERROR",
resources={
"service.name": "test-service",
},
attributes={},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "jsonl",
"source": "logs",
"filter": "severity_text = 'ERROR'",
}
# Export logs with filter
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 2, f"Expected 2 lines (filtered), got {len(jsonl_lines)}"
# Verify only ERROR logs are returned
for line in jsonl_lines:
obj = json.loads(line)
assert obj["severity_text"] == "ERROR"
assert "error message" in obj["body"].lower()
def test_export_logs_with_limit(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert 5 logs.
Tests:
1. Export logs with limit applied
2. Verify only limited number of logs are returned
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
logs = []
for i in range(5):
logs.append(
Logs(
timestamp=now - timedelta(seconds=i),
body=f"Log message {i}",
severity_text="INFO",
resources={
"service.name": "test-service",
},
attributes={
"index": i,
},
)
)
insert_logs(logs)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "csv",
"source": "logs",
"limit": 3,
}
# Export logs with limit
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "text/csv"
# Parse CSV content
csv_content = response.text
csv_reader = csv.DictReader(io.StringIO(csv_content))
rows = list(csv_reader)
assert len(rows) == 3, f"Expected 3 rows (limited), got {len(rows)}"
def test_export_logs_with_columns(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert logs with various attributes.
Tests:
1. Export logs with specific columns
2. Verify only specified columns are returned
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=10),
body="Test log message",
severity_text="INFO",
resources={
"service.name": "test-service",
"deployment.environment": "production",
},
attributes={
"http.method": "GET",
"http.status_code": 200,
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
# Request only specific columns
params = {
"start": start_ns,
"end": end_ns,
"format": "csv",
"source": "logs",
"columns": ["timestamp", "severity_text", "body"],
}
# Export logs with specific columns
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params, doseq=True)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "text/csv"
# Parse CSV content
csv_content = response.text
csv_reader = csv.DictReader(io.StringIO(csv_content))
rows = list(csv_reader)
assert len(rows) == 1
# Verify the specified columns are present
row = rows[0]
assert "timestamp" in row
assert "severity_text" in row
assert "body" in row
assert row["severity_text"] == "INFO"
assert row["body"] == "Test log message"
def test_export_logs_with_order_by(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert logs at different timestamps.
Tests:
1. Export logs with ascending timestamp order
2. Verify logs are returned in correct order
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=10),
body="First log",
severity_text="INFO",
resources={
"service.name": "test-service",
},
attributes={},
),
Logs(
timestamp=now - timedelta(seconds=5),
body="Second log",
severity_text="INFO",
resources={
"service.name": "test-service",
},
attributes={},
),
Logs(
timestamp=now - timedelta(seconds=1),
body="Third log",
severity_text="INFO",
resources={
"service.name": "test-service",
},
attributes={},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "jsonl",
"source": "logs",
"order_by": "timestamp:asc",
}
# Export logs with ascending order
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 3
# Verify order - first log should be "First log" (oldest)
json_objects = [json.loads(line) for line in jsonl_lines]
assert json_objects[0]["body"] == "First log"
assert json_objects[1]["body"] == "Second log"
assert json_objects[2]["body"] == "Third log"
def test_export_logs_with_complex_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[List[Logs]], None],
) -> None:
"""
Setup:
Insert logs with various service names and severity levels.
Tests:
1. Export logs with complex filter (multiple conditions)
2. Verify only logs matching all conditions are returned
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=10),
body="API error occurred",
severity_text="ERROR",
resources={
"service.name": "api-service",
},
attributes={},
),
Logs(
timestamp=now - timedelta(seconds=8),
body="Worker info message",
severity_text="INFO",
resources={
"service.name": "worker-service",
},
attributes={},
),
Logs(
timestamp=now - timedelta(seconds=5),
body="API info message",
severity_text="INFO",
resources={
"service.name": "api-service",
},
attributes={},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
# Filter for api-service AND ERROR severity
params = {
"start": start_ns,
"end": end_ns,
"format": "jsonl",
"source": "logs",
"filter": "service.name = 'api-service' AND severity_text = 'ERROR'",
}
# Export logs with complex filter
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 1, f"Expected 1 line (complex filter), got {len(jsonl_lines)}"
# Verify the filtered log
filtered_obj = json.loads(jsonl_lines[0])
assert filtered_obj["body"] == "API error occurred"
assert filtered_obj["severity_text"] == "ERROR"

View File

@@ -1,773 +0,0 @@
from datetime import datetime, timedelta, timezone
from http import HTTPStatus
from typing import Callable, List
from urllib.parse import urlencode
import csv
import io
import json
import requests
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def test_export_traces_csv(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[List[Traces]], None],
) -> None:
"""
Setup:
Insert 3 traces with different attributes.
Tests:
1. Export traces as CSV format
2. Verify CSV structure and content
3. Validate headers are present
4. Check trace data is correctly formatted
"""
http_service_trace_id = TraceIdGenerator.trace_id()
http_service_span_id = TraceIdGenerator.span_id()
http_service_db_span_id = TraceIdGenerator.span_id()
topic_service_trace_id = TraceIdGenerator.trace_id()
topic_service_span_id = TraceIdGenerator.span_id()
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=3),
trace_id=http_service_trace_id,
span_id=http_service_span_id,
parent_span_id="",
name="POST /integration",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"deployment.environment": "production",
"service.name": "http-service",
"os.type": "linux",
"host.name": "linux-000",
},
attributes={
"net.transport": "IP.TCP",
"http.scheme": "http",
"http.user_agent": "Integration Test",
"http.request.method": "POST",
"http.response.status_code": "200",
},
),
Traces(
timestamp=now - timedelta(seconds=3.5),
duration=timedelta(seconds=0.5),
trace_id=http_service_trace_id,
span_id=http_service_db_span_id,
parent_span_id=http_service_span_id,
name="SELECT",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"deployment.environment": "production",
"service.name": "http-service",
"os.type": "linux",
"host.name": "linux-000",
},
attributes={
"db.name": "integration",
"db.operation": "SELECT",
"db.statement": "SELECT * FROM integration",
},
),
Traces(
timestamp=now - timedelta(seconds=1),
duration=timedelta(seconds=2),
trace_id=topic_service_trace_id,
span_id=topic_service_span_id,
parent_span_id="",
name="topic publish",
kind=TracesKind.SPAN_KIND_PRODUCER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"deployment.environment": "production",
"service.name": "topic-service",
"os.type": "linux",
"host.name": "linux-001",
},
attributes={
"message.type": "SENT",
"messaging.operation": "publish",
"messaging.message.id": "001",
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"source": "traces",
"limit": 1000,
}
# Export traces as CSV (GET for simple queries)
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "text/csv"
assert "attachment" in response.headers.get("Content-Disposition", "")
# Parse CSV content
csv_content = response.text
csv_reader = csv.DictReader(io.StringIO(csv_content))
rows = list(csv_reader)
assert len(rows) == 3, f"Expected 3 rows, got {len(rows)}"
# Verify trace IDs are present in the exported data
trace_ids = [row.get("trace_id") for row in rows]
assert http_service_trace_id in trace_ids
assert topic_service_trace_id in trace_ids
# Verify span names are present
span_names = [row.get("name") for row in rows]
assert "POST /integration" in span_names
assert "SELECT" in span_names
assert "topic publish" in span_names
def test_export_traces_jsonl(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[List[Traces]], None],
) -> None:
"""
Setup:
Insert 2 traces with different attributes.
Tests:
1. Export traces as JSONL format
2. Verify JSONL structure and content
3. Check each line is valid JSON
4. Validate trace data is correctly formatted
"""
http_service_trace_id = TraceIdGenerator.trace_id()
http_service_span_id = TraceIdGenerator.span_id()
topic_service_trace_id = TraceIdGenerator.trace_id()
topic_service_span_id = TraceIdGenerator.span_id()
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=3),
trace_id=http_service_trace_id,
span_id=http_service_span_id,
parent_span_id="",
name="POST /api/test",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "api-service",
"deployment.environment": "staging",
},
attributes={
"http.request.method": "POST",
"http.response.status_code": "201",
},
),
Traces(
timestamp=now - timedelta(seconds=2),
duration=timedelta(seconds=1),
trace_id=topic_service_trace_id,
span_id=topic_service_span_id,
parent_span_id="",
name="queue.process",
kind=TracesKind.SPAN_KIND_CONSUMER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "queue-service",
"deployment.environment": "staging",
},
attributes={
"messaging.operation": "process",
"messaging.system": "rabbitmq",
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "jsonl",
"source": "traces",
"limit": 1000,
}
# Export traces as JSONL (GET for simple queries)
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
assert "attachment" in response.headers.get("Content-Disposition", "")
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 2, f"Expected 2 lines, got {len(jsonl_lines)}"
# Verify each line is valid JSON
json_objects = []
for line in jsonl_lines:
obj = json.loads(line)
json_objects.append(obj)
assert "trace_id" in obj
assert "span_id" in obj
assert "name" in obj
# Verify trace IDs are present
trace_ids = [obj.get("trace_id") for obj in json_objects]
assert http_service_trace_id in trace_ids
assert topic_service_trace_id in trace_ids
# Verify span names are present
span_names = [obj.get("name") for obj in json_objects]
assert "POST /api/test" in span_names
assert "queue.process" in span_names
def test_export_traces_with_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[List[Traces]], None],
) -> None:
"""
Setup:
Insert traces with different service names.
Tests:
1. Export traces with filter applied
2. Verify only filtered traces are returned
"""
service_a_trace_id = TraceIdGenerator.trace_id()
service_a_span_id = TraceIdGenerator.span_id()
service_b_trace_id = TraceIdGenerator.trace_id()
service_b_span_id = TraceIdGenerator.span_id()
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=1),
trace_id=service_a_trace_id,
span_id=service_a_span_id,
parent_span_id="",
name="operation-a",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "service-a",
},
attributes={},
),
Traces(
timestamp=now - timedelta(seconds=2),
duration=timedelta(seconds=1),
trace_id=service_b_trace_id,
span_id=service_b_span_id,
parent_span_id="",
name="operation-b",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "service-b",
},
attributes={},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "jsonl",
"source": "traces",
"limit": 1000,
"filter": "service.name = 'service-a'",
}
# Export traces with filter (GET supports filter param)
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 1, f"Expected 1 line (filtered), got {len(jsonl_lines)}"
# Verify the filtered trace
filtered_obj = json.loads(jsonl_lines[0])
assert filtered_obj["trace_id"] == service_a_trace_id
assert filtered_obj["name"] == "operation-a"
def test_export_traces_with_limit(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[List[Traces]], None],
) -> None:
"""
Setup:
Insert 5 traces.
Tests:
1. Export traces with limit applied
2. Verify only limited number of traces are returned
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
traces = []
for i in range(5):
traces.append(
Traces(
timestamp=now - timedelta(seconds=i),
duration=timedelta(seconds=1),
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
parent_span_id="",
name=f"operation-{i}",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "test-service",
},
attributes={},
)
)
insert_traces(traces)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
params = {
"start": start_ns,
"end": end_ns,
"format": "csv",
"source": "traces",
"limit": 3,
}
# Export traces with limit (GET supports limit param)
response = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/export_raw_data?{urlencode(params)}"),
timeout=10,
headers={
"authorization": f"Bearer {token}",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "text/csv"
# Parse CSV content
csv_content = response.text
csv_reader = csv.DictReader(io.StringIO(csv_content))
rows = list(csv_reader)
assert len(rows) == 3, f"Expected 3 rows (limited), got {len(rows)}"
def test_export_traces_multiple_queries_rejected(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
"""
Tests:
1. POST with multiple builder queries but no trace operator is rejected
2. Verify 400 error is returned
"""
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
body = {
"start": start_ns,
"end": end_ns,
"compositeQuery": {
"queries": [
{
"type": "builder_query",
"spec": {
"signal": "traces",
"name": "A",
"limit": 1000,
"filter": {"expression": "service.name = 'service-a'"},
},
},
{
"type": "builder_query",
"spec": {
"signal": "traces",
"name": "B",
"limit": 1000,
"filter": {"expression": "service.name = 'service-b'"},
},
},
]
},
}
url = signoz.self.host_configs["8080"].get("/api/v1/export_raw_data?format=jsonl")
response = requests.post(
url,
json=body,
timeout=10,
headers={
"authorization": f"Bearer {token}",
"Content-Type": "application/json",
},
)
assert response.status_code == HTTPStatus.BAD_REQUEST
def test_export_traces_with_composite_query_trace_operator(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[List[Traces]], None],
) -> None:
"""
Setup:
Insert multiple traces with parent-child relationships.
Tests:
1. Export traces using trace operator in composite query (POST)
2. Verify trace operator query works correctly
"""
parent_trace_id = TraceIdGenerator.trace_id()
parent_span_id = TraceIdGenerator.span_id()
child_span_id_1 = TraceIdGenerator.span_id()
child_span_id_2 = TraceIdGenerator.span_id()
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=10),
duration=timedelta(seconds=5),
trace_id=parent_trace_id,
span_id=parent_span_id,
parent_span_id="",
name="parent-operation",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "parent-service",
},
attributes={
"operation.type": "parent",
},
),
Traces(
timestamp=now - timedelta(seconds=9),
duration=timedelta(seconds=2),
trace_id=parent_trace_id,
span_id=child_span_id_1,
parent_span_id=parent_span_id,
name="child-operation-1",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "parent-service",
},
attributes={
"operation.type": "child",
},
),
Traces(
timestamp=now - timedelta(seconds=7),
duration=timedelta(seconds=1),
trace_id=parent_trace_id,
span_id=child_span_id_2,
parent_span_id=parent_span_id,
name="child-operation-2",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "parent-service",
},
attributes={
"operation.type": "child",
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
# A: spans with operation.type = 'parent'
query_a = {
"type": "builder_query",
"spec": {
"signal": "traces",
"name": "A",
"limit": 1000,
"filter": {"expression": "operation.type = 'parent'"},
},
}
# B: spans with operation.type = 'child'
query_b = {
"type": "builder_query",
"spec": {
"signal": "traces",
"name": "B",
"limit": 1000,
"filter": {"expression": "operation.type = 'child'"},
},
}
# Trace operator: find traces where A has a direct descendant B
query_c = {
"type": "builder_trace_operator",
"spec": {
"name": "C",
"expression": "A => B",
"returnSpansFrom": "A",
"limit": 1000,
"order": [{"key": {"name": "timestamp"}, "direction": "desc"}]
},
}
body = {
"start": start_ns,
"end": end_ns,
"requestType": "raw",
"compositeQuery": {
"queries": [query_a, query_b, query_c],
},
}
url = signoz.self.host_configs["8080"].get("/api/v1/export_raw_data?format=jsonl")
response = requests.post(
url,
json=body,
timeout=10,
headers={
"authorization": f"Bearer {token}",
"Content-Type": "application/json",
},
)
print(response.text)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 1, f"Expected at least 1 line, got {len(jsonl_lines)}"
# Verify all returned spans belong to the matched trace
json_objects = [json.loads(line) for line in jsonl_lines]
trace_ids = [obj.get("trace_id") for obj in json_objects]
assert all(tid == parent_trace_id for tid in trace_ids)
# Verify the parent span (returnSpansFrom = "A") is present
span_names = [obj.get("name") for obj in json_objects]
assert "parent-operation" in span_names
def test_export_traces_with_select_fields(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[List[Traces]], None],
) -> None:
"""
Setup:
Insert traces with various attributes.
Tests:
1. Export traces with specific select fields via POST
2. Verify only specified fields are returned in the output
"""
trace_id = TraceIdGenerator.trace_id()
span_id = TraceIdGenerator.span_id()
now = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=10),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=span_id,
parent_span_id="",
name="test-operation",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={
"service.name": "test-service",
"deployment.environment": "production",
"host.name": "server-01",
},
attributes={
"http.method": "POST",
"http.status_code": "201",
"user.id": "user123",
},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# Calculate timestamps in nanoseconds
start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
end_ns = int(now.timestamp() * 1e9)
body = {
"start": start_ns,
"end": end_ns,
"requestType": "raw",
"compositeQuery": {
"queries": [
{
"type": "builder_query",
"spec": {
"signal": "traces",
"name": "A",
"limit": 1000,
"selectFields": [
{
"name": "trace_id",
"fieldDataType": "string",
"fieldContext": "span",
"signal": "traces",
},
{
"name": "span_id",
"fieldDataType": "string",
"fieldContext": "span",
"signal": "traces",
},
{
"name": "name",
"fieldDataType": "string",
"fieldContext": "span",
"signal": "traces",
},
{
"name": "service.name",
"fieldDataType": "string",
"fieldContext": "resource",
"signal": "traces",
},
],
},
}
]
},
}
url = signoz.self.host_configs["8080"].get("/api/v1/export_raw_data?format=jsonl")
response = requests.post(
url,
json=body,
timeout=10,
headers={
"authorization": f"Bearer {token}",
"Content-Type": "application/json",
},
)
assert response.status_code == HTTPStatus.OK
assert response.headers["Content-Type"] == "application/x-ndjson"
# Parse JSONL content
jsonl_lines = response.text.strip().split("\n")
assert len(jsonl_lines) == 1
# Verify the selected fields are present
result = json.loads(jsonl_lines[0])
assert "trace_id" in result
assert "span_id" in result
assert "name" in result
# Verify values
assert result["trace_id"] == trace_id
assert result["span_id"] == span_id
assert result["name"] == "test-operation"