refactor: code clean up

.github/workflows/integrationci.yaml

@@ -55,8 +55,6 @@ jobs:
         sqlstore-provider:
           - postgres
           - sqlite
-        sqlite-mode:
-          - wal
         clickhouse-version:
           - 25.5.6
           - 25.12.5
@@ -104,7 +102,6 @@ jobs:
             --basetemp=./tmp/ \
             src/${{matrix.src}} \
             --sqlstore-provider ${{matrix.sqlstore-provider}} \
-            --sqlite-mode ${{matrix.sqlite-mode}} \
             --postgres-version ${{matrix.postgres-version}} \
             --clickhouse-version ${{matrix.clickhouse-version}} \
             --schema-migrator-version ${{matrix.schema-migrator-version}}

cmd/community/server.go

@@ -17,11 +17,15 @@ import (
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/gateway"
 	"github.com/SigNoz/signoz/pkg/gateway/noopgateway"
+	"github.com/SigNoz/signoz/pkg/global"
 	"github.com/SigNoz/signoz/pkg/licensing"
 	"github.com/SigNoz/signoz/pkg/licensing/nooplicensing"
+	"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
+	"github.com/SigNoz/signoz/pkg/modules/cloudintegration/implcloudintegration"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard/impldashboard"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
+	"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/query-service/app"
 	"github.com/SigNoz/signoz/pkg/queryparser"
@@ -29,6 +33,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlschema"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
 	"github.com/SigNoz/signoz/pkg/version"
 	"github.com/SigNoz/signoz/pkg/zeus"
 	"github.com/SigNoz/signoz/pkg/zeus/noopzeus"
@@ -96,6 +101,9 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 		func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
 			return querier.NewHandler(ps, q, a)
 		},
+		func(_ cloudintegrationtypes.Store, _ global.Global, _ zeus.Zeus, _ gateway.Gateway, _ licensing.Licensing, _ serviceaccount.Module, _ cloudintegration.Config) (cloudintegration.Module, error) {
+			return implcloudintegration.NewModule(), nil
+		},
 	)
 	if err != nil {
 		logger.ErrorContext(ctx, "failed to create signoz", errors.Attr(err))

cmd/enterprise/server.go

@@ -16,6 +16,8 @@ import (
 	"github.com/SigNoz/signoz/ee/gateway/httpgateway"
 	enterpriselicensing "github.com/SigNoz/signoz/ee/licensing"
 	"github.com/SigNoz/signoz/ee/licensing/httplicensing"
+	"github.com/SigNoz/signoz/ee/modules/cloudintegration/implcloudintegration"
+	"github.com/SigNoz/signoz/ee/modules/cloudintegration/implcloudintegration/implcloudprovider"
 	"github.com/SigNoz/signoz/ee/modules/dashboard/impldashboard"
 	eequerier "github.com/SigNoz/signoz/ee/querier"
 	enterpriseapp "github.com/SigNoz/signoz/ee/query-service/app"
@@ -29,10 +31,14 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/gateway"
+	"github.com/SigNoz/signoz/pkg/global"
 	"github.com/SigNoz/signoz/pkg/licensing"
+	"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
+	pkgcloudintegration "github.com/SigNoz/signoz/pkg/modules/cloudintegration/implcloudintegration"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard"
 	pkgimpldashboard "github.com/SigNoz/signoz/pkg/modules/dashboard/impldashboard"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
+	"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/queryparser"
 	"github.com/SigNoz/signoz/pkg/signoz"
@@ -40,6 +46,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/sqlstore/sqlstorehook"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
 	"github.com/SigNoz/signoz/pkg/version"
 	"github.com/SigNoz/signoz/pkg/zeus"
 )
@@ -125,7 +132,6 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 				return nil, err
 			}
 			return openfgaauthz.NewProviderFactory(sqlstore, openfgaschema.NewSchema().Get(ctx), openfgaDataStore, licensing, dashboardModule), nil
-
 		},
 		func(store sqlstore.SQLStore, settings factory.ProviderSettings, analytics analytics.Analytics, orgGetter organization.Getter, queryParser queryparser.QueryParser, querier querier.Querier, licensing licensing.Licensing) dashboard.Module {
 			return impldashboard.NewModule(pkgimpldashboard.NewStore(store), settings, analytics, orgGetter, queryParser, querier, licensing)
@@ -137,8 +143,21 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 			communityHandler := querier.NewHandler(ps, q, a)
 			return eequerier.NewHandler(ps, q, communityHandler)
 		},
-	)
+		func(store cloudintegrationtypes.Store, global global.Global, zeus zeus.Zeus, gateway gateway.Gateway, licensing licensing.Licensing, serviceAccount serviceaccount.Module, config cloudintegration.Config) (cloudintegration.Module, error) {
+			defStore := pkgcloudintegration.NewServiceDefinitionStore()
+			awsCloudProviderModule, err := implcloudprovider.NewAWSCloudProvider(defStore)
+			if err != nil {
+				return nil, err
+			}
+			azureCloudProviderModule := implcloudprovider.NewAzureCloudProvider()
+			cloudProvidersMap := map[cloudintegrationtypes.CloudProviderType]cloudintegration.CloudProviderModule{
+				cloudintegrationtypes.CloudProviderTypeAWS:   awsCloudProviderModule,
+				cloudintegrationtypes.CloudProviderTypeAzure: azureCloudProviderModule,
+			}
 
+			return implcloudintegration.NewModule(store, global, zeus, gateway, licensing, serviceAccount, cloudProvidersMap, config)
+		},
+	)
 	if err != nil {
 		logger.ErrorContext(ctx, "failed to create signoz", errors.Attr(err))
 		return err

conf/example.yaml

@@ -86,7 +86,7 @@ sqlstore:
     # The path to the SQLite database file.
     path: /var/lib/signoz/signoz.db
     # The journal mode for the sqlite database. Supported values: delete, wal.
-    mode: wal
+    mode: delete
     # The timeout for the sqlite database to wait for a lock.
     busy_timeout: 10s
     # The default transaction locking behavior. Supported values: deferred, immediate, exclusive.
@@ -364,3 +364,10 @@ serviceaccount:
   analytics:
     # toggle service account analytics
     enabled: true
+
+##################### Cloud Integration #####################
+cloudintegration:
+  # cloud integration agent configuration
+  agent:
+    # The version of the cloud integration agent.
+    version: v0.0.8

docs/api/openapi.yml

@@ -327,6 +327,27 @@ components:
           nullable: true
           type: array
       type: object
+    AuthtypesStorableRole:
+      properties:
+        createdAt:
+          format: date-time
+          type: string
+        description:
+          type: string
+        id:
+          type: string
+        name:
+          type: string
+        orgId:
+          type: string
+        type:
+          type: string
+        updatedAt:
+          format: date-time
+          type: string
+      required:
+      - id
+      type: object
     AuthtypesTransaction:
       properties:
         object:
@@ -350,7 +371,7 @@ components:
         id:
           type: string
         role:
-          $ref: '#/components/schemas/AuthtypesRole'
+          $ref: '#/components/schemas/AuthtypesStorableRole'
         roleId:
           type: string
         updatedAt:
@@ -360,11 +381,6 @@ components:
           type: string
       required:
       - id
-      - userId
-      - roleId
-      - createdAt
-      - updatedAt
-      - role
       type: object
     AuthtypesUserWithRoles:
       properties:
@@ -405,11 +421,11 @@ components:
       type: object
     CloudintegrationtypesAWSCollectionStrategy:
       properties:
-        aws_logs:
+        logs:
           $ref: '#/components/schemas/CloudintegrationtypesAWSLogsStrategy'
-        aws_metrics:
+        metrics:
           $ref: '#/components/schemas/CloudintegrationtypesAWSMetricsStrategy'
-        s3_buckets:
+        s3Buckets:
           additionalProperties:
             items:
               type: string
@@ -449,12 +465,12 @@ components:
       type: object
     CloudintegrationtypesAWSLogsStrategy:
       properties:
-        cloudwatch_logs_subscriptions:
+        cloudwatchLogsSubscriptions:
           items:
             properties:
-              filter_pattern:
+              filterPattern:
                 type: string
-              log_group_name_prefix:
+              logGroupNamePrefix:
                 type: string
             type: object
           nullable: true
@@ -462,7 +478,7 @@ components:
       type: object
     CloudintegrationtypesAWSMetricsStrategy:
       properties:
-        cloudwatch_metric_stream_filters:
+        cloudwatchMetricStreamFilters:
           items:
             properties:
               MetricNames:
@@ -486,7 +502,7 @@ components:
       properties:
         enabled:
           type: boolean
-        s3_buckets:
+        s3Buckets:
           additionalProperties:
             items:
               type: string
@@ -561,6 +577,26 @@ components:
           nullable: true
           type: array
       type: object
+    CloudintegrationtypesCloudIntegrationService:
+      nullable: true
+      properties:
+        cloudIntegrationId:
+          type: string
+        config:
+          $ref: '#/components/schemas/CloudintegrationtypesServiceConfig'
+        createdAt:
+          format: date-time
+          type: string
+        id:
+          type: string
+        type:
+          $ref: '#/components/schemas/CloudintegrationtypesServiceID'
+        updatedAt:
+          format: date-time
+          type: string
+      required:
+      - id
+      type: object
     CloudintegrationtypesCollectedLogAttribute:
       properties:
         name:
@@ -596,6 +632,16 @@ components:
       - aws
       type: object
     CloudintegrationtypesConnectionArtifactRequest:
+      properties:
+        config:
+          $ref: '#/components/schemas/CloudintegrationtypesConnectionArtifactRequestConfig'
+        credentials:
+          $ref: '#/components/schemas/CloudintegrationtypesSignozCredentials'
+      required:
+      - config
+      - credentials
+      type: object
+    CloudintegrationtypesConnectionArtifactRequestConfig:
       properties:
         aws:
           $ref: '#/components/schemas/CloudintegrationtypesAWSConnectionArtifactRequest'
@@ -694,11 +740,54 @@ components:
             type: string
           type: array
         telemetry:
-          $ref: '#/components/schemas/CloudintegrationtypesAWSCollectionStrategy'
+          $ref: '#/components/schemas/CloudintegrationtypesOldAWSCollectionStrategy'
       required:
       - enabled_regions
       - telemetry
       type: object
+    CloudintegrationtypesOldAWSCollectionStrategy:
+      properties:
+        aws_logs:
+          $ref: '#/components/schemas/CloudintegrationtypesOldAWSLogsStrategy'
+        aws_metrics:
+          $ref: '#/components/schemas/CloudintegrationtypesOldAWSMetricsStrategy'
+        provider:
+          type: string
+        s3_buckets:
+          additionalProperties:
+            items:
+              type: string
+            type: array
+          type: object
+      type: object
+    CloudintegrationtypesOldAWSLogsStrategy:
+      properties:
+        cloudwatch_logs_subscriptions:
+          items:
+            properties:
+              filter_pattern:
+                type: string
+              log_group_name_prefix:
+                type: string
+            type: object
+          nullable: true
+          type: array
+      type: object
+    CloudintegrationtypesOldAWSMetricsStrategy:
+      properties:
+        cloudwatch_metric_stream_filters:
+          items:
+            properties:
+              MetricNames:
+                items:
+                  type: string
+                type: array
+              Namespace:
+                type: string
+            type: object
+          nullable: true
+          type: array
+      type: object
     CloudintegrationtypesPostableAgentCheckInRequest:
       properties:
         account_id:
@@ -727,6 +816,8 @@ components:
       properties:
         assets:
           $ref: '#/components/schemas/CloudintegrationtypesAssets'
+        cloudIntegrationService:
+          $ref: '#/components/schemas/CloudintegrationtypesCloudIntegrationService'
         dataCollected:
           $ref: '#/components/schemas/CloudintegrationtypesDataCollected'
         icon:
@@ -735,9 +826,7 @@ components:
           type: string
         overview:
           type: string
-        serviceConfig:
-          $ref: '#/components/schemas/CloudintegrationtypesServiceConfig'
-        supported_signals:
+        supportedSignals:
           $ref: '#/components/schemas/CloudintegrationtypesSupportedSignals'
         telemetryCollectionStrategy:
           $ref: '#/components/schemas/CloudintegrationtypesCollectionStrategy'
@@ -749,9 +838,10 @@ components:
       - icon
       - overview
       - assets
-      - supported_signals
+      - supportedSignals
       - dataCollected
       - telemetryCollectionStrategy
+      - cloudIntegrationService
       type: object
     CloudintegrationtypesServiceConfig:
       properties:
@@ -760,6 +850,22 @@ components:
       required:
       - aws
       type: object
+    CloudintegrationtypesServiceID:
+      enum:
+      - alb
+      - api-gateway
+      - dynamodb
+      - ec2
+      - ecs
+      - eks
+      - elasticache
+      - lambda
+      - msk
+      - rds
+      - s3sync
+      - sns
+      - sqs
+      type: string
     CloudintegrationtypesServiceMetadata:
       properties:
         enabled:
@@ -776,6 +882,22 @@ components:
       - icon
       - enabled
       type: object
+    CloudintegrationtypesSignozCredentials:
+      properties:
+        ingestionKey:
+          type: string
+        ingestionUrl:
+          type: string
+        sigNozApiKey:
+          type: string
+        sigNozApiURL:
+          type: string
+      required:
+      - sigNozApiURL
+      - sigNozApiKey
+      - ingestionUrl
+      - ingestionKey
+      type: object
     CloudintegrationtypesSupportedSignals:
       properties:
         logs:
@@ -2297,6 +2419,15 @@ components:
       - status
       - error
       type: object
+    RulestatehistorytypesAlertState:
+      enum:
+      - inactive
+      - pending
+      - recovering
+      - firing
+      - nodata
+      - disabled
+      type: string
     RulestatehistorytypesGettableRuleStateHistory:
       properties:
         fingerprint:
@@ -2308,15 +2439,15 @@ components:
           nullable: true
           type: array
         overallState:
-          $ref: '#/components/schemas/RuletypesAlertState'
+          $ref: '#/components/schemas/RulestatehistorytypesAlertState'
         overallStateChanged:
           type: boolean
-        ruleId:
+        ruleID:
           type: string
         ruleName:
           type: string
         state:
-          $ref: '#/components/schemas/RuletypesAlertState'
+          $ref: '#/components/schemas/RulestatehistorytypesAlertState'
         stateChanged:
           type: boolean
         unixMilli:
@@ -2326,7 +2457,7 @@ components:
           format: double
           type: number
       required:
-      - ruleId
+      - ruleID
       - ruleName
       - overallState
       - overallStateChanged
@@ -2416,21 +2547,12 @@ components:
           format: int64
           type: integer
         state:
-          $ref: '#/components/schemas/RuletypesAlertState'
+          $ref: '#/components/schemas/RulestatehistorytypesAlertState'
       required:
       - state
       - start
       - end
       type: object
-    RuletypesAlertState:
-      enum:
-      - inactive
-      - pending
-      - recovering
-      - firing
-      - nodata
-      - disabled
-      type: string
     ServiceaccounttypesGettableFactorAPIKey:
       properties:
         createdAt:
@@ -3394,6 +3516,61 @@ paths:
       summary: Update account
       tags:
       - cloudintegration
+  /api/v1/cloud_integrations/{cloud_provider}/accounts/{id}/services/{service_id}:
+    put:
+      deprecated: false
+      description: This endpoint updates a service for the specified cloud provider
+      operationId: UpdateService
+      parameters:
+      - in: path
+        name: cloud_provider
+        required: true
+        schema:
+          type: string
+      - in: path
+        name: id
+        required: true
+        schema:
+          type: string
+      - in: path
+        name: service_id
+        required: true
+        schema:
+          type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CloudintegrationtypesUpdatableService'
+      responses:
+        "204":
+          description: No Content
+        "401":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Unauthorized
+        "403":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Forbidden
+        "500":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Internal Server Error
+      security:
+      - api_key:
+        - ADMIN
+      - tokenizer:
+        - ADMIN
+      summary: Update service
+      tags:
+      - cloudintegration
   /api/v1/cloud_integrations/{cloud_provider}/accounts/check_in:
     post:
       deprecated: false
@@ -3451,6 +3628,59 @@ paths:
       summary: Agent check-in
       tags:
       - cloudintegration
+  /api/v1/cloud_integrations/{cloud_provider}/credentials:
+    get:
+      deprecated: false
+      description: This endpoint retrieves the connection credentials required for
+        integration
+      operationId: GetConnectionCredentials
+      parameters:
+      - in: path
+        name: cloud_provider
+        required: true
+        schema:
+          type: string
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                properties:
+                  data:
+                    $ref: '#/components/schemas/CloudintegrationtypesSignozCredentials'
+                  status:
+                    type: string
+                required:
+                - status
+                - data
+                type: object
+          description: OK
+        "401":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Unauthorized
+        "403":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Forbidden
+        "500":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/RenderErrorResponse'
+          description: Internal Server Error
+      security:
+      - api_key:
+        - ADMIN
+      - tokenizer:
+        - ADMIN
+      summary: Get connection credentials
+      tags:
+      - cloudintegration
   /api/v1/cloud_integrations/{cloud_provider}/services:
     get:
       deprecated: false
@@ -3561,55 +3791,6 @@ paths:
       summary: Get service
       tags:
       - cloudintegration
-    put:
-      deprecated: false
-      description: This endpoint updates a service for the specified cloud provider
-      operationId: UpdateService
-      parameters:
-      - in: path
-        name: cloud_provider
-        required: true
-        schema:
-          type: string
-      - in: path
-        name: service_id
-        required: true
-        schema:
-          type: string
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: '#/components/schemas/CloudintegrationtypesUpdatableService'
-      responses:
-        "204":
-          description: No Content
-        "401":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Unauthorized
-        "403":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Forbidden
-        "500":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Internal Server Error
-      security:
-      - api_key:
-        - ADMIN
-      - tokenizer:
-        - ADMIN
-      summary: Update service
-      tags:
-      - cloudintegration
   /api/v1/complete/google:
     get:
       deprecated: false
@@ -8453,7 +8634,7 @@ paths:
       - in: query
         name: state
         schema:
-          $ref: '#/components/schemas/RuletypesAlertState'
+          $ref: '#/components/schemas/RulestatehistorytypesAlertState'
       - in: query
         name: filterExpression
         schema:

docs/contributing/go/integration.md

@@ -193,7 +193,6 @@ uv run pytest --basetemp=./tmp/ -vv --reuse src/passwordauthn/01_register.py::te
 Tests can be configured using pytest options:
 
 - `--sqlstore-provider` - Choose database provider (default: postgres)
-- `--sqlite-mode` - SQLite journal mode: `delete` or `wal` (default: delete). Only relevant when `--sqlstore-provider=sqlite`.
 - `--postgres-version` - PostgreSQL version (default: 15)
 - `--clickhouse-version` - ClickHouse version (default: 25.5.6)
 - `--zookeeper-version` - Zookeeper version (default: 3.7.1)
@@ -203,6 +202,7 @@ Example:
 uv run pytest --basetemp=./tmp/ -vv --reuse --sqlstore-provider=postgres --postgres-version=14 src/auth/
 ```
 
+
 ## What should I remember?
 
 - **Always use the `--reuse` flag** when setting up the environment to keep containers running
@@ -213,4 +213,3 @@ uv run pytest --basetemp=./tmp/ -vv --reuse --sqlstore-provider=postgres --postg
 - **Use descriptive test names** that clearly indicate what is being tested
 - **Leverage fixtures** for common setup and authentication
 - **Test both success and failure scenarios** to ensure robust functionality
-- **`--sqlite-mode=wal` does not work on macOS.** The integration test environment runs SigNoz inside a Linux container with the SQLite database file mounted from the macOS host. WAL mode requires shared memory between connections, and connections crossing the VM boundary (macOS host ↔ Linux container) cannot share the WAL index, resulting in `SQLITE_IOERR_SHORT_READ`. WAL mode is tested in CI on Linux only.

ee/anomaly/params.go

@@ -32,7 +32,7 @@ func (s Seasonality) IsValid() bool {
 }
 
 type AnomaliesRequest struct {
-	Params      *qbtypes.QueryRangeRequest
+	Params      qbtypes.QueryRangeRequest
 	Seasonality Seasonality
 }
 
@@ -81,7 +81,7 @@ type anomalyQueryParams struct {
 	Past3SeasonQuery qbtypes.QueryRangeRequest
 }
 
-func prepareAnomalyQueryParams(req *qbtypes.QueryRangeRequest, seasonality Seasonality) *anomalyQueryParams {
+func prepareAnomalyQueryParams(req qbtypes.QueryRangeRequest, seasonality Seasonality) *anomalyQueryParams {
 	start := req.Start
 	end := req.End
 

ee/modules/cloudintegration/implcloudintegration/implcloudprovider/awscloudprovider.go

@@ -0,0 +1,132 @@
+package implcloudprovider
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"sort"
+
+	"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
+	"github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
+)
+
+type awscloudprovider struct {
+	serviceDefinitions cloudintegrationtypes.ServiceDefinitionStore
+}
+
+func NewAWSCloudProvider(defStore cloudintegrationtypes.ServiceDefinitionStore) (cloudintegration.CloudProviderModule, error) {
+	return &awscloudprovider{serviceDefinitions: defStore}, nil
+}
+
+func (provider *awscloudprovider) GetConnectionArtifact(ctx context.Context, account *cloudintegrationtypes.Account, req *cloudintegrationtypes.ConnectionArtifactRequest) (*cloudintegrationtypes.ConnectionArtifact, error) {
+	baseURL := fmt.Sprintf(cloudintegrationtypes.CloudFormationQuickCreateBaseURL.StringValue(), req.Config.Aws.DeploymentRegion)
+	u, _ := url.Parse(baseURL)
+
+	q := u.Query()
+	q.Set("region", req.Config.Aws.DeploymentRegion)
+	u.Fragment = "/stacks/quickcreate"
+
+	u.RawQuery = q.Encode()
+
+	q = u.Query()
+	q.Set("stackName", cloudintegrationtypes.AgentCloudFormationBaseStackName.StringValue())
+	q.Set("templateURL", fmt.Sprintf(cloudintegrationtypes.AgentCloudFormationTemplateS3Path.StringValue(), req.Config.AgentVersion))
+	q.Set("param_SigNozIntegrationAgentVersion", req.Config.AgentVersion)
+	q.Set("param_SigNozApiUrl", req.Credentials.SigNozAPIURL)
+	q.Set("param_SigNozApiKey", req.Credentials.SigNozAPIKey)
+	q.Set("param_SigNozAccountId", account.ID.StringValue())
+	q.Set("param_IngestionUrl", req.Credentials.IngestionURL)
+	q.Set("param_IngestionKey", req.Credentials.IngestionKey)
+
+	return &cloudintegrationtypes.ConnectionArtifact{
+		Aws: &cloudintegrationtypes.AWSConnectionArtifact{
+			ConnectionURL: u.String() + "?&" + q.Encode(), // this format is required by AWS
+		},
+	}, nil
+}
+
+func (provider *awscloudprovider) ListServiceDefinitions(ctx context.Context) ([]*cloudintegrationtypes.ServiceDefinition, error) {
+	return provider.serviceDefinitions.List(ctx, cloudintegrationtypes.CloudProviderTypeAWS)
+}
+
+func (provider *awscloudprovider) GetServiceDefinition(ctx context.Context, serviceID cloudintegrationtypes.ServiceID) (*cloudintegrationtypes.ServiceDefinition, error) {
+	serviceDef, err := provider.serviceDefinitions.Get(ctx, cloudintegrationtypes.CloudProviderTypeAWS, serviceID)
+	if err != nil {
+		return nil, err
+	}
+
+	// override cloud integration dashboard id
+	for index, dashboard := range serviceDef.Assets.Dashboards {
+		serviceDef.Assets.Dashboards[index].ID = cloudintegrationtypes.GetCloudIntegrationDashboardID(cloudintegrationtypes.CloudProviderTypeAWS, serviceID.StringValue(), dashboard.ID)
+	}
+
+	return serviceDef, nil
+}
+
+func (provider *awscloudprovider) BuildIntegrationConfig(
+	ctx context.Context,
+	account *cloudintegrationtypes.Account,
+	services []*cloudintegrationtypes.StorableCloudIntegrationService,
+) (*cloudintegrationtypes.ProviderIntegrationConfig, error) {
+	// Sort services for deterministic output
+	sort.Slice(services, func(i, j int) bool {
+		return services[i].Type.StringValue() < services[j].Type.StringValue()
+	})
+
+	compiledMetrics := new(cloudintegrationtypes.AWSMetricsStrategy)
+	compiledLogs := new(cloudintegrationtypes.AWSLogsStrategy)
+	var compiledS3Buckets map[string][]string
+
+	for _, storedSvc := range services {
+		svcCfg, err := cloudintegrationtypes.NewServiceConfigFromJSON(cloudintegrationtypes.CloudProviderTypeAWS, storedSvc.Config)
+		if err != nil {
+			return nil, err
+		}
+
+		svcDef, err := provider.GetServiceDefinition(ctx, storedSvc.Type)
+		if err != nil {
+			return nil, err
+		}
+
+		strategy := svcDef.Strategy.AWS
+		logsEnabled := cloudintegrationtypes.IsLogsEnabled(cloudintegrationtypes.CloudProviderTypeAWS, svcCfg)
+
+		// S3Sync: logs come directly from configured S3 buckets, not CloudWatch subscriptions
+		if storedSvc.Type == cloudintegrationtypes.AWSServiceS3Sync {
+			if logsEnabled && svcCfg.AWS.Logs.S3Buckets != nil {
+				compiledS3Buckets = svcCfg.AWS.Logs.S3Buckets
+			}
+			// no need to go ahead as the code block specifically checks for the S3Sync service
+			continue
+		}
+
+		if logsEnabled && strategy.Logs != nil {
+			compiledLogs.Subscriptions = append(compiledLogs.Subscriptions, strategy.Logs.Subscriptions...)
+		}
+
+		metricsEnabled := cloudintegrationtypes.IsMetricsEnabled(cloudintegrationtypes.CloudProviderTypeAWS, svcCfg)
+
+		if metricsEnabled && strategy.Metrics != nil {
+			compiledMetrics.StreamFilters = append(compiledMetrics.StreamFilters, strategy.Metrics.StreamFilters...)
+		}
+	}
+
+	awsTelemetry := new(cloudintegrationtypes.AWSCollectionStrategy)
+
+	if len(compiledMetrics.StreamFilters) > 0 {
+		awsTelemetry.Metrics = compiledMetrics
+	}
+	if len(compiledLogs.Subscriptions) > 0 {
+		awsTelemetry.Logs = compiledLogs
+	}
+	if compiledS3Buckets != nil {
+		awsTelemetry.S3Buckets = compiledS3Buckets
+	}
+
+	return &cloudintegrationtypes.ProviderIntegrationConfig{
+		AWS: &cloudintegrationtypes.AWSIntegrationConfig{
+			EnabledRegions: account.Config.AWS.Regions,
+			Telemetry:      awsTelemetry,
+		},
+	}, nil
+}

ee/modules/cloudintegration/implcloudintegration/implcloudprovider/azurecloudprovider.go

@@ -0,0 +1,34 @@
+package implcloudprovider
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
+	"github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
+)
+
+type azurecloudprovider struct{}
+
+func NewAzureCloudProvider() cloudintegration.CloudProviderModule {
+	return &azurecloudprovider{}
+}
+
+func (provider *azurecloudprovider) GetConnectionArtifact(ctx context.Context, account *cloudintegrationtypes.Account, req *cloudintegrationtypes.ConnectionArtifactRequest) (*cloudintegrationtypes.ConnectionArtifact, error) {
+	panic("implement me")
+}
+
+func (provider *azurecloudprovider) ListServiceDefinitions(ctx context.Context) ([]*cloudintegrationtypes.ServiceDefinition, error) {
+	panic("implement me")
+}
+
+func (provider *azurecloudprovider) GetServiceDefinition(ctx context.Context, serviceID cloudintegrationtypes.ServiceID) (*cloudintegrationtypes.ServiceDefinition, error) {
+	panic("implement me")
+}
+
+func (provider *azurecloudprovider) BuildIntegrationConfig(
+	ctx context.Context,
+	account *cloudintegrationtypes.Account,
+	services []*cloudintegrationtypes.StorableCloudIntegrationService,
+) (*cloudintegrationtypes.ProviderIntegrationConfig, error) {
+	panic("implement me")
+}

ee/modules/cloudintegration/implcloudintegration/module.go

@@ -0,0 +1,513 @@
+package implcloudintegration
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/gateway"
+	"github.com/SigNoz/signoz/pkg/global"
+	"github.com/SigNoz/signoz/pkg/licensing"
+	"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
+	"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
+	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
+	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
+	"github.com/SigNoz/signoz/pkg/types/serviceaccounttypes"
+	"github.com/SigNoz/signoz/pkg/types/zeustypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/SigNoz/signoz/pkg/zeus"
+)
+
+type module struct {
+	store             cloudintegrationtypes.Store
+	gateway           gateway.Gateway
+	zeus              zeus.Zeus
+	licensing         licensing.Licensing
+	global            global.Global
+	serviceAccount    serviceaccount.Module
+	cloudProvidersMap map[cloudintegrationtypes.CloudProviderType]cloudintegration.CloudProviderModule
+	config            cloudintegration.Config
+}
+
+func NewModule(
+	store cloudintegrationtypes.Store,
+	global global.Global,
+	zeus zeus.Zeus,
+	gateway gateway.Gateway,
+	licensing licensing.Licensing,
+	serviceAccount serviceaccount.Module,
+	cloudProvidersMap map[cloudintegrationtypes.CloudProviderType]cloudintegration.CloudProviderModule,
+	config cloudintegration.Config,
+) (cloudintegration.Module, error) {
+	return &module{
+		store:             store,
+		global:            global,
+		zeus:              zeus,
+		gateway:           gateway,
+		licensing:         licensing,
+		serviceAccount:    serviceAccount,
+		cloudProvidersMap: cloudProvidersMap,
+		config:            config,
+	}, nil
+}
+
+// GetConnectionCredentials returns credentials required to generate connection artifact. eg. apiKey, ingestionKey etc.
+// It will return creds it can deduce and return empty value for others.
+func (module *module) GetConnectionCredentials(ctx context.Context, orgID valuer.UUID, provider cloudintegrationtypes.CloudProviderType) (*cloudintegrationtypes.SignozCredentials, error) {
+	// get license to get the deployment details
+	license, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	var ingestionURL string
+
+	globalConfig := module.global.GetConfig(ctx)
+	if globalConfig != nil {
+		ingestionURL = globalConfig.IngestionURL
+	}
+
+	// get deployment details from zeus, ignore error
+	respBytes, _ := module.zeus.GetDeployment(ctx, license.Key)
+
+	var signozAPIURL string
+
+	if len(respBytes) > 0 {
+		// parse deployment details, ignore error, if client is asking api url every time check for possible error
+		deployment, _ := zeustypes.NewGettableDeployment(respBytes)
+		if deployment != nil {
+			signozAPIURL = deployment.SignozAPIUrl
+		}
+	}
+
+	// ignore error
+	apiKey, _ := module.getOrCreateAPIKey(ctx, orgID, provider)
+
+	// ignore error
+	ingestionKey, _ := module.getOrCreateIngestionKey(ctx, orgID, provider)
+
+	return &cloudintegrationtypes.SignozCredentials{
+		SigNozAPIURL: signozAPIURL,
+		SigNozAPIKey: apiKey,
+		IngestionURL: ingestionURL,
+		IngestionKey: ingestionKey,
+	}, nil
+}
+
+func (module *module) CreateAccount(ctx context.Context, account *cloudintegrationtypes.Account) error {
+	_, err := module.licensing.GetActive(ctx, account.OrgID)
+	if err != nil {
+		return errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	storableCloudIntegration, err := cloudintegrationtypes.NewStorableCloudIntegration(account)
+	if err != nil {
+		return err
+	}
+
+	return module.store.CreateAccount(ctx, storableCloudIntegration)
+}
+
+func (module *module) GetConnectionArtifact(ctx context.Context, account *cloudintegrationtypes.Account, req *cloudintegrationtypes.ConnectionArtifactRequest) (*cloudintegrationtypes.ConnectionArtifact, error) {
+	cloudProviderModule, err := module.getCloudProvider(account.Provider)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Config.AddAgentVersion(module.config.Agent.Version)
+
+	return cloudProviderModule.GetConnectionArtifact(ctx, account, req)
+}
+
+func (module *module) GetAccount(ctx context.Context, orgID valuer.UUID, accountID valuer.UUID, provider cloudintegrationtypes.CloudProviderType) (*cloudintegrationtypes.Account, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	storableAccount, err := module.store.GetAccountByID(ctx, orgID, accountID, provider)
+	if err != nil {
+		return nil, err
+	}
+
+	return cloudintegrationtypes.NewAccountFromStorable(storableAccount)
+}
+
+// ListAccounts return only agent connected accounts.
+func (module *module) ListAccounts(ctx context.Context, orgID valuer.UUID, provider cloudintegrationtypes.CloudProviderType) ([]*cloudintegrationtypes.Account, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	storableAccounts, err := module.store.ListConnectedAccounts(ctx, orgID, provider)
+	if err != nil {
+		return nil, err
+	}
+
+	return cloudintegrationtypes.NewAccountsFromStorables(storableAccounts)
+}
+
+func (module *module) AgentCheckIn(ctx context.Context, orgID valuer.UUID, provider cloudintegrationtypes.CloudProviderType, req *cloudintegrationtypes.AgentCheckInRequest) (*cloudintegrationtypes.AgentCheckInResponse, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	connectedAccount, err := module.store.GetConnectedAccount(ctx, orgID, provider, req.ProviderAccountID)
+	if err != nil && !errors.Ast(err, errors.TypeNotFound) {
+		return nil, err
+	}
+
+	// If a different integration is already connected to this provider account ID, reject the check-in.
+	// Allow re-check-in from the same integration (e.g. agent restarting).
+	if connectedAccount != nil && connectedAccount.ID != req.CloudIntegrationID {
+		errMessage := fmt.Sprintf("provider account id %s is already connected to cloud integration id %s", req.ProviderAccountID, connectedAccount.ID)
+		return nil, errors.New(errors.TypeAlreadyExists, cloudintegrationtypes.ErrCodeCloudIntegrationAlreadyConnected, errMessage)
+	}
+
+	account, err := module.store.GetAccountByID(ctx, orgID, req.CloudIntegrationID, provider)
+	if err != nil {
+		return nil, err
+	}
+
+	// update account with cloud provider account id and agent report (heartbeat)
+	account.Update(&req.ProviderAccountID, cloudintegrationtypes.NewAgentReport(req.Data))
+
+	err = module.store.UpdateAccount(ctx, account)
+	if err != nil {
+		return nil, err
+	}
+
+	// If account has been removed (disconnected), return a minimal response with empty integration config.
+	// The agent doesn't act on config for removed accounts.
+	if account.RemovedAt != nil {
+		return &cloudintegrationtypes.AgentCheckInResponse{
+			CloudIntegrationID: account.ID.StringValue(),
+			ProviderAccountID:  req.ProviderAccountID,
+			IntegrationConfig:  new(cloudintegrationtypes.ProviderIntegrationConfig),
+			RemovedAt:          account.RemovedAt,
+		}, nil
+	}
+
+	// Get account as domain object for config access (enabled regions, etc.)
+	domainAccount, err := cloudintegrationtypes.NewAccountFromStorable(account)
+	if err != nil {
+		return nil, err
+	}
+
+	cloudProvider, err := module.getCloudProvider(provider)
+	if err != nil {
+		return nil, err
+	}
+
+	storedServices, err := module.store.ListServices(ctx, req.CloudIntegrationID)
+	if err != nil {
+		return nil, err
+	}
+
+	// Delegate integration config building entirely to the provider module
+	integrationConfig, err := cloudProvider.BuildIntegrationConfig(ctx, domainAccount, storedServices)
+	if err != nil {
+		return nil, err
+	}
+
+	return &cloudintegrationtypes.AgentCheckInResponse{
+		CloudIntegrationID: account.ID.StringValue(),
+		ProviderAccountID:  req.ProviderAccountID,
+		IntegrationConfig:  integrationConfig,
+		RemovedAt:          account.RemovedAt,
+	}, nil
+}
+
+func (module *module) UpdateAccount(ctx context.Context, account *cloudintegrationtypes.Account) error {
+	_, err := module.licensing.GetActive(ctx, account.OrgID)
+	if err != nil {
+		return errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	storableAccount, err := cloudintegrationtypes.NewStorableCloudIntegration(account)
+	if err != nil {
+		return err
+	}
+
+	return module.store.UpdateAccount(ctx, storableAccount)
+}
+
+func (module *module) DisconnectAccount(ctx context.Context, orgID valuer.UUID, accountID valuer.UUID, provider cloudintegrationtypes.CloudProviderType) error {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	return module.store.RemoveAccount(ctx, orgID, accountID, provider)
+}
+
+func (module *module) ListServicesMetadata(ctx context.Context, orgID valuer.UUID, provider cloudintegrationtypes.CloudProviderType, integrationID *valuer.UUID) ([]*cloudintegrationtypes.ServiceMetadata, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	cloudProvider, err := module.getCloudProvider(provider)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceDefinitions, err := cloudProvider.ListServiceDefinitions(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	enabledServiceIDs := map[string]bool{}
+	if integrationID != nil {
+		storedServices, err := module.store.ListServices(ctx, *integrationID)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, svc := range storedServices {
+			serviceConfig, err := cloudintegrationtypes.NewServiceConfigFromJSON(provider, svc.Config)
+			if err != nil {
+				return nil, err
+			}
+
+			if cloudintegrationtypes.IsServiceEnabled(provider, serviceConfig) {
+				enabledServiceIDs[svc.Type.StringValue()] = true
+			}
+		}
+	}
+
+	resp := make([]*cloudintegrationtypes.ServiceMetadata, 0, len(serviceDefinitions))
+	for _, serviceDefinition := range serviceDefinitions {
+		resp = append(resp, cloudintegrationtypes.NewServiceMetadata(*serviceDefinition, enabledServiceIDs[serviceDefinition.ID]))
+	}
+
+	return resp, nil
+}
+
+func (module *module) GetService(ctx context.Context, orgID valuer.UUID, integrationID *valuer.UUID, serviceID cloudintegrationtypes.ServiceID, provider cloudintegrationtypes.CloudProviderType) (*cloudintegrationtypes.Service, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	cloudProvider, err := module.getCloudProvider(provider)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceDefinition, err := cloudProvider.GetServiceDefinition(ctx, serviceID)
+	if err != nil {
+		return nil, err
+	}
+
+	var integrationService *cloudintegrationtypes.CloudIntegrationService
+
+	if integrationID != nil {
+		storedService, err := module.store.GetServiceByServiceID(ctx, *integrationID, serviceID)
+		if err != nil && !errors.Ast(err, errors.TypeNotFound) {
+			return nil, err
+		}
+		if storedService != nil {
+			serviceConfig, err := cloudintegrationtypes.NewServiceConfigFromJSON(provider, storedService.Config)
+			if err != nil {
+				return nil, err
+			}
+
+			integrationService = cloudintegrationtypes.NewCloudIntegrationServiceFromStorable(storedService, serviceConfig)
+		}
+	}
+
+	return cloudintegrationtypes.NewService(*serviceDefinition, integrationService), nil
+}
+
+func (module *module) CreateService(ctx context.Context, orgID valuer.UUID, service *cloudintegrationtypes.CloudIntegrationService, provider cloudintegrationtypes.CloudProviderType) error {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	cloudProvider, err := module.getCloudProvider(provider)
+	if err != nil {
+		return err
+	}
+
+	serviceDefinition, err := cloudProvider.GetServiceDefinition(ctx, service.Type)
+	if err != nil {
+		return err
+	}
+
+	configJSON, err := service.Config.ToJSON(provider, service.Type, &serviceDefinition.SupportedSignals)
+	if err != nil {
+		return err
+	}
+
+	return module.store.CreateService(ctx, cloudintegrationtypes.NewStorableCloudIntegrationService(service, string(configJSON)))
+}
+
+func (module *module) UpdateService(ctx context.Context, orgID valuer.UUID, integrationService *cloudintegrationtypes.CloudIntegrationService, provider cloudintegrationtypes.CloudProviderType) error {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	cloudProvider, err := module.getCloudProvider(provider)
+	if err != nil {
+		return err
+	}
+
+	serviceDefinition, err := cloudProvider.GetServiceDefinition(ctx, integrationService.Type)
+	if err != nil {
+		return err
+	}
+
+	configJSON, err := integrationService.Config.ToJSON(provider, integrationService.Type, &serviceDefinition.SupportedSignals)
+	if err != nil {
+		return err
+	}
+
+	storableService := cloudintegrationtypes.NewStorableCloudIntegrationService(integrationService, string(configJSON))
+
+	return module.store.UpdateService(ctx, storableService)
+}
+
+func (module *module) GetDashboardByID(ctx context.Context, orgID valuer.UUID, id string) (*dashboardtypes.Dashboard, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	_, _, _, err = cloudintegrationtypes.ParseCloudIntegrationDashboardID(id)
+	if err != nil {
+		return nil, err
+	}
+
+	allDashboards, err := module.listDashboards(ctx, orgID)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, d := range allDashboards {
+		if d.ID == id {
+			return d, nil
+		}
+	}
+
+	return nil, errors.New(errors.TypeNotFound, cloudintegrationtypes.ErrCodeCloudIntegrationNotFound, "cloud integration dashboard not found")
+}
+
+func (module *module) ListDashboards(ctx context.Context, orgID valuer.UUID) ([]*dashboardtypes.Dashboard, error) {
+	_, err := module.licensing.GetActive(ctx, orgID)
+	if err != nil {
+		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
+	}
+
+	return module.listDashboards(ctx, orgID)
+}
+
+func (module *module) getCloudProvider(provider cloudintegrationtypes.CloudProviderType) (cloudintegration.CloudProviderModule, error) {
+	if cloudProviderModule, ok := module.cloudProvidersMap[provider]; ok {
+		return cloudProviderModule, nil
+	}
+
+	return nil, errors.NewInvalidInputf(cloudintegrationtypes.ErrCodeCloudProviderInvalidInput, "invalid cloud provider: %s", provider.StringValue())
+}
+
+func (module *module) getOrCreateIngestionKey(ctx context.Context, orgID valuer.UUID, provider cloudintegrationtypes.CloudProviderType) (string, error) {
+	keyName := cloudintegrationtypes.NewIngestionKeyName(provider)
+
+	result, err := module.gateway.SearchIngestionKeysByName(ctx, orgID, keyName, 1, 10)
+	if err != nil {
+		return "", errors.WrapInternalf(err, errors.CodeInternal, "couldn't search ingestion keys")
+	}
+
+	// ideally there should be only one key per cloud integration provider
+	if len(result.Keys) > 0 {
+		return result.Keys[0].Value, nil
+	}
+
+	createdIngestionKey, err := module.gateway.CreateIngestionKey(ctx, orgID, keyName, []string{"integration"}, time.Time{})
+	if err != nil {
+		return "", errors.WrapInternalf(err, errors.CodeInternal, "couldn't create ingestion key")
+	}
+
+	return createdIngestionKey.Value, nil
+}
+
+func (module *module) getOrCreateAPIKey(ctx context.Context, orgID valuer.UUID, provider cloudintegrationtypes.CloudProviderType) (string, error) {
+	domain := module.serviceAccount.Config().Email.Domain
+	serviceAccount := serviceaccounttypes.NewServiceAccount("integration", domain, serviceaccounttypes.ServiceAccountStatusActive, orgID)
+	serviceAccount, err := module.serviceAccount.GetOrCreate(ctx, orgID, serviceAccount)
+	if err != nil {
+		return "", err
+	}
+	err = module.serviceAccount.SetRoleByName(ctx, orgID, serviceAccount.ID, authtypes.SigNozViewerRoleName)
+	if err != nil {
+		return "", err
+	}
+
+	factorAPIKey, err := serviceAccount.NewFactorAPIKey(provider.StringValue(), 0)
+	if err != nil {
+		return "", err
+	}
+
+	factorAPIKey, err = module.serviceAccount.GetOrCreateFactorAPIKey(ctx, factorAPIKey)
+	if err != nil {
+		return "", err
+	}
+	return factorAPIKey.Key, nil
+}
+
+func (module *module) listDashboards(ctx context.Context, orgID valuer.UUID) ([]*dashboardtypes.Dashboard, error) {
+	var allDashboards []*dashboardtypes.Dashboard
+
+	for provider := range module.cloudProvidersMap {
+		cloudProvider, err := module.getCloudProvider(provider)
+		if err != nil {
+			return nil, err
+		}
+
+		connectedAccounts, err := module.store.ListConnectedAccounts(ctx, orgID, provider)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, storableAccount := range connectedAccounts {
+			storedServices, err := module.store.ListServices(ctx, storableAccount.ID)
+			if err != nil {
+				return nil, err
+			}
+
+			for _, storedSvc := range storedServices {
+				serviceConfig, err := cloudintegrationtypes.NewServiceConfigFromJSON(provider, storedSvc.Config)
+				if err != nil || !cloudintegrationtypes.IsMetricsEnabled(provider, serviceConfig) {
+					continue
+				}
+
+				svcDef, err := cloudProvider.GetServiceDefinition(ctx, storedSvc.Type)
+				if err != nil || svcDef == nil {
+					continue
+				}
+
+				dashboards := cloudintegrationtypes.GetDashboardsFromAssets(
+					storedSvc.Type.StringValue(),
+					orgID,
+					provider,
+					storableAccount.CreatedAt,
+					svcDef.Assets,
+				)
+				allDashboards = append(allDashboards, dashboards...)
+			}
+		}
+	}
+
+	sort.Slice(allDashboards, func(i, j int) bool {
+		return allDashboards[i].ID < allDashboards[j].ID
+	})
+
+	return allDashboards, nil
+}

ee/querier/handler.go

@@ -65,7 +65,7 @@ func (h *handler) QueryRange(rw http.ResponseWriter, req *http.Request) {
 	}
 
 	if anomalyQuery, ok := queryRangeRequest.IsAnomalyRequest(); ok {
-		anomalies, err := h.handleAnomalyQuery(ctx, orgID, anomalyQuery, &queryRangeRequest)
+		anomalies, err := h.handleAnomalyQuery(ctx, orgID, anomalyQuery, queryRangeRequest)
 		if err != nil {
 			render.Error(rw, errors.NewInternalf(errors.CodeInternal, "failed to get anomalies: %v", err))
 			return
@@ -149,7 +149,7 @@ func (h *handler) createAnomalyProvider(seasonality anomalyV2.Seasonality) anoma
 	}
 }
 
-func (h *handler) handleAnomalyQuery(ctx context.Context, orgID valuer.UUID, anomalyQuery *qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation], queryRangeRequest *qbtypes.QueryRangeRequest) (*anomalyV2.AnomaliesResponse, error) {
+func (h *handler) handleAnomalyQuery(ctx context.Context, orgID valuer.UUID, anomalyQuery *qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation], queryRangeRequest qbtypes.QueryRangeRequest) (*anomalyV2.AnomaliesResponse, error) {
 	seasonality := extractSeasonality(anomalyQuery)
 	provider := h.createAnomalyProvider(seasonality)
 

ee/query-service/app/server.go

@@ -49,6 +49,7 @@ import (
 	opAmpModel "github.com/SigNoz/signoz/pkg/query-service/app/opamp/model"
 	baseconst "github.com/SigNoz/signoz/pkg/query-service/constants"
 	"github.com/SigNoz/signoz/pkg/query-service/healthcheck"
+	baseint "github.com/SigNoz/signoz/pkg/query-service/interfaces"
 	baserules "github.com/SigNoz/signoz/pkg/query-service/rules"
 	"github.com/SigNoz/signoz/pkg/query-service/utils"
 )
@@ -98,6 +99,7 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 	)
 
 	rm, err := makeRulesManager(
+		reader,
 		signoz.Cache,
 		signoz.Alertmanager,
 		signoz.SQLStore,
@@ -343,7 +345,7 @@ func (s *Server) Stop(ctx context.Context) error {
 	return nil
 }
 
-func makeRulesManager(cache cache.Cache, alertmanager alertmanager.Alertmanager, sqlstore sqlstore.SQLStore, telemetryStore telemetrystore.TelemetryStore, metadataStore telemetrytypes.MetadataStore, prometheus prometheus.Prometheus, orgGetter organization.Getter, ruleStateHistoryModule rulestatehistory.Module, querier querier.Querier, providerSettings factory.ProviderSettings, queryParser queryparser.QueryParser) (*baserules.Manager, error) {
+func makeRulesManager(ch baseint.Reader, cache cache.Cache, alertmanager alertmanager.Alertmanager, sqlstore sqlstore.SQLStore, telemetryStore telemetrystore.TelemetryStore, metadataStore telemetrytypes.MetadataStore, prometheus prometheus.Prometheus, orgGetter organization.Getter, ruleStateHistoryModule rulestatehistory.Module, querier querier.Querier, providerSettings factory.ProviderSettings, queryParser queryparser.QueryParser) (*baserules.Manager, error) {
 	ruleStore := sqlrulestore.NewRuleStore(sqlstore, queryParser, providerSettings)
 	maintenanceStore := sqlrulestore.NewMaintenanceStore(sqlstore)
 	// create manager opts
@@ -352,6 +354,7 @@ func makeRulesManager(cache cache.Cache, alertmanager alertmanager.Alertmanager,
 		MetadataStore:          metadataStore,
 		Prometheus:             prometheus,
 		Context:                context.Background(),
+		Reader:                 ch,
 		Querier:                querier,
 		Logger:                 providerSettings.Logger,
 		Cache:                  cache,
@@ -362,7 +365,7 @@ func makeRulesManager(cache cache.Cache, alertmanager alertmanager.Alertmanager,
 		OrgGetter:              orgGetter,
 		RuleStore:              ruleStore,
 		MaintenanceStore:       maintenanceStore,
-		SQLStore:               sqlstore,
+		SqlStore:               sqlstore,
 		QueryParser:            queryParser,
 		RuleStateHistoryModule: ruleStateHistoryModule,
 	}

ee/query-service/rules/anomaly.go

@@ -5,34 +5,58 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"math"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/SigNoz/signoz/ee/query-service/anomaly"
+	"github.com/SigNoz/signoz/pkg/cache"
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/querier"
-	"github.com/SigNoz/signoz/pkg/types/rulestatehistorytypes"
+	"github.com/SigNoz/signoz/pkg/query-service/common"
+	"github.com/SigNoz/signoz/pkg/query-service/model"
+	"github.com/SigNoz/signoz/pkg/transition"
 	"github.com/SigNoz/signoz/pkg/types/ruletypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 
+	querierV2 "github.com/SigNoz/signoz/pkg/query-service/app/querier/v2"
+	"github.com/SigNoz/signoz/pkg/query-service/app/queryBuilder"
+	"github.com/SigNoz/signoz/pkg/query-service/interfaces"
+	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
+	"github.com/SigNoz/signoz/pkg/query-service/utils/labels"
+	"github.com/SigNoz/signoz/pkg/query-service/utils/times"
+	"github.com/SigNoz/signoz/pkg/query-service/utils/timestamp"
+
 	"github.com/SigNoz/signoz/pkg/units"
 
 	baserules "github.com/SigNoz/signoz/pkg/query-service/rules"
 
-	"github.com/SigNoz/signoz/ee/anomaly"
+	querierV5 "github.com/SigNoz/signoz/pkg/querier"
+
+	anomalyV2 "github.com/SigNoz/signoz/ee/anomaly"
 
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 )
 
+const (
+	RuleTypeAnomaly = "anomaly_rule"
+)
+
 type AnomalyRule struct {
 	*baserules.BaseRule
 
 	mtx sync.Mutex
 
-	// querier is used for alerts migrated after the introduction of new query builder
-	querier querier.Querier
+	reader interfaces.Reader
 
-	provider anomaly.Provider
+	// querierV2 is used for alerts created after the introduction of new metrics query builder
+	querierV2 interfaces.Querier
+
+	// querierV5 is used for alerts migrated after the introduction of new query builder
+	querierV5 querierV5.Querier
+
+	provider   anomaly.Provider
+	providerV2 anomalyV2.Provider
 
 	version string
 	logger  *slog.Logger
@@ -46,16 +70,18 @@ func NewAnomalyRule(
 	id string,
 	orgID valuer.UUID,
 	p *ruletypes.PostableRule,
-	querier querier.Querier,
+	reader interfaces.Reader,
+	querierV5 querierV5.Querier,
 	logger *slog.Logger,
+	cache cache.Cache,
 	opts ...baserules.RuleOption,
 ) (*AnomalyRule, error) {
 
-	logger.Info("creating new AnomalyRule", slog.String("rule.id", id))
+	logger.Info("creating new AnomalyRule", "rule_id", id)
 
 	opts = append(opts, baserules.WithLogger(logger))
 
-	baseRule, err := baserules.NewBaseRule(id, orgID, p, opts...)
+	baseRule, err := baserules.NewBaseRule(id, orgID, p, reader, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -75,38 +101,93 @@ func NewAnomalyRule(
 		t.seasonality = anomaly.SeasonalityDaily
 	}
 
-	logger.Info("using seasonality", slog.String("rule.id", id), slog.String("rule.seasonality", t.seasonality.StringValue()))
+	logger.Info("using seasonality", "seasonality", t.seasonality.String())
 
+	querierOptsV2 := querierV2.QuerierOptions{
+		Reader:       reader,
+		Cache:        cache,
+		KeyGenerator: queryBuilder.NewKeyGenerator(),
+	}
+
+	t.querierV2 = querierV2.NewQuerier(querierOptsV2)
+	t.reader = reader
 	if t.seasonality == anomaly.SeasonalityHourly {
 		t.provider = anomaly.NewHourlyProvider(
-			anomaly.WithQuerier[*anomaly.HourlyProvider](querier),
-			anomaly.WithLogger[*anomaly.HourlyProvider](logger),
+			anomaly.WithCache[*anomaly.HourlyProvider](cache),
+			anomaly.WithKeyGenerator[*anomaly.HourlyProvider](queryBuilder.NewKeyGenerator()),
+			anomaly.WithReader[*anomaly.HourlyProvider](reader),
 		)
 	} else if t.seasonality == anomaly.SeasonalityDaily {
 		t.provider = anomaly.NewDailyProvider(
-			anomaly.WithQuerier[*anomaly.DailyProvider](querier),
-			anomaly.WithLogger[*anomaly.DailyProvider](logger),
+			anomaly.WithCache[*anomaly.DailyProvider](cache),
+			anomaly.WithKeyGenerator[*anomaly.DailyProvider](queryBuilder.NewKeyGenerator()),
+			anomaly.WithReader[*anomaly.DailyProvider](reader),
 		)
 	} else if t.seasonality == anomaly.SeasonalityWeekly {
 		t.provider = anomaly.NewWeeklyProvider(
-			anomaly.WithQuerier[*anomaly.WeeklyProvider](querier),
-			anomaly.WithLogger[*anomaly.WeeklyProvider](logger),
+			anomaly.WithCache[*anomaly.WeeklyProvider](cache),
+			anomaly.WithKeyGenerator[*anomaly.WeeklyProvider](queryBuilder.NewKeyGenerator()),
+			anomaly.WithReader[*anomaly.WeeklyProvider](reader),
 		)
 	}
 
-	t.querier = querier
+	if t.seasonality == anomaly.SeasonalityHourly {
+		t.providerV2 = anomalyV2.NewHourlyProvider(
+			anomalyV2.WithQuerier[*anomalyV2.HourlyProvider](querierV5),
+			anomalyV2.WithLogger[*anomalyV2.HourlyProvider](logger),
+		)
+	} else if t.seasonality == anomaly.SeasonalityDaily {
+		t.providerV2 = anomalyV2.NewDailyProvider(
+			anomalyV2.WithQuerier[*anomalyV2.DailyProvider](querierV5),
+			anomalyV2.WithLogger[*anomalyV2.DailyProvider](logger),
+		)
+	} else if t.seasonality == anomaly.SeasonalityWeekly {
+		t.providerV2 = anomalyV2.NewWeeklyProvider(
+			anomalyV2.WithQuerier[*anomalyV2.WeeklyProvider](querierV5),
+			anomalyV2.WithLogger[*anomalyV2.WeeklyProvider](logger),
+		)
+	}
+
+	t.querierV5 = querierV5
 	t.version = p.Version
 	t.logger = logger
 	return &t, nil
 }
 
 func (r *AnomalyRule) Type() ruletypes.RuleType {
-	return ruletypes.RuleTypeAnomaly
+	return RuleTypeAnomaly
 }
 
-func (r *AnomalyRule) prepareQueryRange(ctx context.Context, ts time.Time) *qbtypes.QueryRangeRequest {
+func (r *AnomalyRule) prepareQueryRange(ctx context.Context, ts time.Time) (*v3.QueryRangeParamsV3, error) {
 
-	r.logger.InfoContext(ctx, "prepare query range request", slog.String("rule.id", r.ID()), slog.Int64("ts", ts.UnixMilli()), slog.Int64("eval.window_ms", r.EvalWindow().Milliseconds()), slog.Int64("eval.delay_ms", r.EvalDelay().Milliseconds()))
+	r.logger.InfoContext(
+		ctx, "prepare query range request v4", "ts", ts.UnixMilli(), "eval_window", r.EvalWindow().Milliseconds(), "eval_delay", r.EvalDelay().Milliseconds(),
+	)
+
+	st, en := r.Timestamps(ts)
+	start := st.UnixMilli()
+	end := en.UnixMilli()
+
+	compositeQuery := r.Condition().CompositeQuery
+
+	if compositeQuery.PanelType != v3.PanelTypeGraph {
+		compositeQuery.PanelType = v3.PanelTypeGraph
+	}
+
+	// default mode
+	return &v3.QueryRangeParamsV3{
+		Start:          start,
+		End:            end,
+		Step:           int64(math.Max(float64(common.MinAllowedStepInterval(start, end)), 60)),
+		CompositeQuery: compositeQuery,
+		Variables:      make(map[string]interface{}, 0),
+		NoCache:        false,
+	}, nil
+}
+
+func (r *AnomalyRule) prepareQueryRangeV5(ctx context.Context, ts time.Time) (*qbtypes.QueryRangeRequest, error) {
+
+	r.logger.InfoContext(ctx, "prepare query range request v5", "ts", ts.UnixMilli(), "eval_window", r.EvalWindow().Milliseconds(), "eval_delay", r.EvalDelay().Milliseconds())
 
 	startTs, endTs := r.Timestamps(ts)
 	start, end := startTs.UnixMilli(), endTs.UnixMilli()
@@ -122,14 +203,25 @@ func (r *AnomalyRule) prepareQueryRange(ctx context.Context, ts time.Time) *qbty
 	}
 	req.CompositeQuery.Queries = make([]qbtypes.QueryEnvelope, len(r.Condition().CompositeQuery.Queries))
 	copy(req.CompositeQuery.Queries, r.Condition().CompositeQuery.Queries)
-	return req
+	return req, nil
+}
+
+func (r *AnomalyRule) GetSelectedQuery() string {
+	return r.Condition().GetSelectedQueryName()
 }
 
 func (r *AnomalyRule) buildAndRunQuery(ctx context.Context, orgID valuer.UUID, ts time.Time) (ruletypes.Vector, error) {
 
-	params := r.prepareQueryRange(ctx, ts)
+	params, err := r.prepareQueryRange(ctx, ts)
+	if err != nil {
+		return nil, err
+	}
+	err = r.PopulateTemporality(ctx, orgID, params)
+	if err != nil {
+		return nil, fmt.Errorf("internal error while setting temporality")
+	}
 
-	anomalies, err := r.provider.GetAnomalies(ctx, orgID, &anomaly.AnomaliesRequest{
+	anomalies, err := r.provider.GetAnomalies(ctx, orgID, &anomaly.GetAnomaliesRequest{
 		Params:      params,
 		Seasonality: r.seasonality,
 	})
@@ -137,43 +229,87 @@ func (r *AnomalyRule) buildAndRunQuery(ctx context.Context, orgID valuer.UUID, t
 		return nil, err
 	}
 
-	var queryResult *qbtypes.TimeSeriesData
+	var queryResult *v3.Result
 	for _, result := range anomalies.Results {
-		if result.QueryName == r.SelectedQuery(ctx) {
+		if result.QueryName == r.GetSelectedQuery() {
 			queryResult = result
 			break
 		}
 	}
 
-	if queryResult == nil {
-		r.logger.WarnContext(ctx, "nil qb result", slog.String("rule.id", r.ID()), slog.Int64("ts", ts.UnixMilli()))
-		return ruletypes.Vector{}, nil
-	}
-
-	hasData := len(queryResult.Aggregations) > 0 &&
-		queryResult.Aggregations[0] != nil &&
-		len(queryResult.Aggregations[0].AnomalyScores) > 0
-
+	hasData := len(queryResult.AnomalyScores) > 0
 	if missingDataAlert := r.HandleMissingDataAlert(ctx, ts, hasData); missingDataAlert != nil {
 		return ruletypes.Vector{*missingDataAlert}, nil
-	} else if !hasData {
-		r.logger.WarnContext(ctx, "no anomaly result", slog.String("rule.id", r.ID()))
-		return ruletypes.Vector{}, nil
 	}
 
 	var resultVector ruletypes.Vector
 
-	scoresJSON, _ := json.Marshal(queryResult.Aggregations[0].AnomalyScores)
-	// TODO(srikanthccv): this could be noisy but we do this to answer false alert requests
-	r.logger.InfoContext(ctx, "anomaly scores", slog.String("rule.id", r.ID()), slog.String("anomaly.scores", string(scoresJSON)))
+	scoresJSON, _ := json.Marshal(queryResult.AnomalyScores)
+	r.logger.InfoContext(ctx, "anomaly scores", "scores", string(scoresJSON))
+
+	for _, series := range queryResult.AnomalyScores {
+		if !r.Condition().ShouldEval(series) {
+			r.logger.InfoContext(ctx, "not enough data points to evaluate series, skipping", "ruleid", r.ID(), "numPoints", len(series.Points), "requiredPoints", r.Condition().RequiredNumPoints)
+			continue
+		}
+		results, err := r.Threshold.Eval(*series, r.Unit(), ruletypes.EvalData{
+			ActiveAlerts:  r.ActiveAlertsLabelFP(),
+			SendUnmatched: r.ShouldSendUnmatched(),
+		})
+		if err != nil {
+			return nil, err
+		}
+		resultVector = append(resultVector, results...)
+	}
+	return resultVector, nil
+}
+
+func (r *AnomalyRule) buildAndRunQueryV5(ctx context.Context, orgID valuer.UUID, ts time.Time) (ruletypes.Vector, error) {
+
+	params, err := r.prepareQueryRangeV5(ctx, ts)
+	if err != nil {
+		return nil, err
+	}
+
+	anomalies, err := r.providerV2.GetAnomalies(ctx, orgID, &anomalyV2.AnomaliesRequest{
+		Params:      *params,
+		Seasonality: anomalyV2.Seasonality{String: valuer.NewString(r.seasonality.String())},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	var qbResult *qbtypes.TimeSeriesData
+	for _, result := range anomalies.Results {
+		if result.QueryName == r.GetSelectedQuery() {
+			qbResult = result
+			break
+		}
+	}
+
+	if qbResult == nil {
+		r.logger.WarnContext(ctx, "nil qb result", "ts", ts.UnixMilli())
+	}
+
+	queryResult := transition.ConvertV5TimeSeriesDataToV4Result(qbResult)
+
+	hasData := len(queryResult.AnomalyScores) > 0
+	if missingDataAlert := r.HandleMissingDataAlert(ctx, ts, hasData); missingDataAlert != nil {
+		return ruletypes.Vector{*missingDataAlert}, nil
+	}
+
+	var resultVector ruletypes.Vector
+
+	scoresJSON, _ := json.Marshal(queryResult.AnomalyScores)
+	r.logger.InfoContext(ctx, "anomaly scores", "scores", string(scoresJSON))
 
 	// Filter out new series if newGroupEvalDelay is configured
-	seriesToProcess := queryResult.Aggregations[0].AnomalyScores
+	seriesToProcess := queryResult.AnomalyScores
 	if r.ShouldSkipNewGroups() {
 		filteredSeries, filterErr := r.BaseRule.FilterNewSeries(ctx, ts, seriesToProcess)
 		// In case of error we log the error and continue with the original series
 		if filterErr != nil {
-			r.logger.ErrorContext(ctx, "error filtering new series", slog.String("rule.id", r.ID()), errors.Attr(filterErr))
+			r.logger.ErrorContext(ctx, "Error filtering new series, ", errors.Attr(filterErr), "rule_name", r.Name())
 		} else {
 			seriesToProcess = filteredSeries
 		}
@@ -181,10 +317,10 @@ func (r *AnomalyRule) buildAndRunQuery(ctx context.Context, orgID valuer.UUID, t
 
 	for _, series := range seriesToProcess {
 		if !r.Condition().ShouldEval(series) {
-			r.logger.InfoContext(ctx, "not enough data points to evaluate series, skipping", slog.String("rule.id", r.ID()), slog.Int("series.num_points", len(series.Values)), slog.Int("series.required_points", r.Condition().RequiredNumPoints))
+			r.logger.InfoContext(ctx, "not enough data points to evaluate series, skipping", "ruleid", r.ID(), "numPoints", len(series.Points), "requiredPoints", r.Condition().RequiredNumPoints)
 			continue
 		}
-		results, err := r.Threshold.Eval(series, r.Unit(), ruletypes.EvalData{
+		results, err := r.Threshold.Eval(*series, r.Unit(), ruletypes.EvalData{
 			ActiveAlerts:  r.ActiveAlertsLabelFP(),
 			SendUnmatched: r.ShouldSendUnmatched(),
 		})
@@ -205,9 +341,13 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
 	var res ruletypes.Vector
 	var err error
 
-	r.logger.InfoContext(ctx, "running query", slog.String("rule.id", r.ID()))
-	res, err = r.buildAndRunQuery(ctx, r.OrgID(), ts)
-
+	if r.version == "v5" {
+		r.logger.InfoContext(ctx, "running v5 query")
+		res, err = r.buildAndRunQueryV5(ctx, r.OrgID(), ts)
+	} else {
+		r.logger.InfoContext(ctx, "running v4 query")
+		res, err = r.buildAndRunQuery(ctx, r.OrgID(), ts)
+	}
 	if err != nil {
 		return 0, err
 	}
@@ -231,7 +371,7 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
 		}
 		value := valueFormatter.Format(smpl.V, r.Unit())
 		threshold := valueFormatter.Format(smpl.Target, smpl.TargetUnit)
-		r.logger.DebugContext(ctx, "alert template data for rule", slog.String("rule.id", r.ID()), slog.String("formatter.name", valueFormatter.Name()), slog.String("alert.value", value), slog.String("alert.threshold", threshold))
+		r.logger.DebugContext(ctx, "Alert template data for rule", "rule_name", r.Name(), "formatter", valueFormatter.Name(), "value", value, "threshold", threshold)
 
 		tmplData := ruletypes.AlertTemplateData(l, value, threshold)
 		// Inject some convenience variables that are easier to remember for users
@@ -246,34 +386,35 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
 				defs+text,
 				"__alert_"+r.Name(),
 				tmplData,
+				times.Time(timestamp.FromTime(ts)),
 				nil,
 			)
 			result, err := tmpl.Expand()
 			if err != nil {
 				result = fmt.Sprintf("<error expanding template: %s>", err)
-				r.logger.ErrorContext(ctx, "expanding alert template failed", slog.String("rule.id", r.ID()), errors.Attr(err), slog.Any("alert.template_data", tmplData))
+				r.logger.ErrorContext(ctx, "Expanding alert template failed", errors.Attr(err), "data", tmplData, "rule_name", r.Name())
 			}
 			return result
 		}
 
-		lb := ruletypes.NewBuilder(smpl.Metric...).Del(ruletypes.MetricNameLabel).Del(ruletypes.TemporalityLabel)
-		resultLabels := ruletypes.NewBuilder(smpl.Metric...).Del(ruletypes.MetricNameLabel).Del(ruletypes.TemporalityLabel).Labels()
+		lb := labels.NewBuilder(smpl.Metric).Del(labels.MetricNameLabel).Del(labels.TemporalityLabel)
+		resultLabels := labels.NewBuilder(smpl.Metric).Del(labels.MetricNameLabel).Del(labels.TemporalityLabel).Labels()
 
 		for name, value := range r.Labels().Map() {
 			lb.Set(name, expand(value))
 		}
 
-		lb.Set(ruletypes.AlertNameLabel, r.Name())
-		lb.Set(ruletypes.AlertRuleIDLabel, r.ID())
-		lb.Set(ruletypes.RuleSourceLabel, r.GeneratorURL())
+		lb.Set(labels.AlertNameLabel, r.Name())
+		lb.Set(labels.AlertRuleIdLabel, r.ID())
+		lb.Set(labels.RuleSourceLabel, r.GeneratorURL())
 
-		annotations := make(ruletypes.Labels, 0, len(r.Annotations().Map()))
+		annotations := make(labels.Labels, 0, len(r.Annotations().Map()))
 		for name, value := range r.Annotations().Map() {
-			annotations = append(annotations, ruletypes.Label{Name: name, Value: expand(value)})
+			annotations = append(annotations, labels.Label{Name: name, Value: expand(value)})
 		}
 		if smpl.IsMissing {
-			lb.Set(ruletypes.AlertNameLabel, "[No data] "+r.Name())
-			lb.Set(ruletypes.NoDataLabel, "true")
+			lb.Set(labels.AlertNameLabel, "[No data] "+r.Name())
+			lb.Set(labels.NoDataLabel, "true")
 		}
 
 		lbs := lb.Labels()
@@ -281,17 +422,17 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
 		resultFPs[h] = struct{}{}
 
 		if _, ok := alerts[h]; ok {
-			r.logger.ErrorContext(ctx, "the alert query returns duplicate records", slog.String("rule.id", r.ID()), slog.Any("alert", alerts[h]))
-			err = errors.NewInternalf(errors.CodeInternal, "duplicate alert found, vector contains metrics with the same labelset after applying alert labels")
+			r.logger.ErrorContext(ctx, "the alert query returns duplicate records", "rule_id", r.ID(), "alert", alerts[h])
+			err = fmt.Errorf("duplicate alert found, vector contains metrics with the same labelset after applying alert labels")
 			return 0, err
 		}
 
 		alerts[h] = &ruletypes.Alert{
 			Labels:            lbs,
-			QueryResultLabels: resultLabels,
+			QueryResultLables: resultLabels,
 			Annotations:       annotations,
 			ActiveAt:          ts,
-			State:             ruletypes.StatePending,
+			State:             model.StatePending,
 			Value:             smpl.V,
 			GeneratorURL:      r.GeneratorURL(),
 			Receivers:         ruleReceiverMap[lbs.Map()[ruletypes.LabelThresholdName]],
@@ -300,12 +441,12 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
 		}
 	}
 
-	r.logger.InfoContext(ctx, "number of alerts found", slog.String("rule.id", r.ID()), slog.Int("alert.count", len(alerts)))
+	r.logger.InfoContext(ctx, "number of alerts found", "rule_name", r.Name(), "alerts_count", len(alerts))
 	// alerts[h] is ready, add or update active list now
 	for h, a := range alerts {
 		// Check whether we already have alerting state for the identifying label set.
 		// Update the last value and annotations if so, create a new alert entry otherwise.
-		if alert, ok := r.Active[h]; ok && alert.State != ruletypes.StateInactive {
+		if alert, ok := r.Active[h]; ok && alert.State != model.StateInactive {
 
 			alert.Value = a.Value
 			alert.Annotations = a.Annotations
@@ -321,76 +462,76 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
 		r.Active[h] = a
 	}
 
-	itemsToAdd := []rulestatehistorytypes.RuleStateHistory{}
+	itemsToAdd := []model.RuleStateHistory{}
 
 	// Check if any pending alerts should be removed or fire now. Write out alert timeseries.
 	for fp, a := range r.Active {
-		labelsJSON, err := json.Marshal(a.QueryResultLabels)
+		labelsJSON, err := json.Marshal(a.QueryResultLables)
 		if err != nil {
-			r.logger.ErrorContext(ctx, "error marshaling labels", slog.String("rule.id", r.ID()), errors.Attr(err), slog.Any("alert.labels", a.Labels))
+			r.logger.ErrorContext(ctx, "error marshaling labels", errors.Attr(err), "labels", a.Labels)
 		}
 		if _, ok := resultFPs[fp]; !ok {
 			// If the alert was previously firing, keep it around for a given
 			// retention time so it is reported as resolved to the AlertManager.
-			if a.State == ruletypes.StatePending || (!a.ResolvedAt.IsZero() && ts.Sub(a.ResolvedAt) > ruletypes.ResolvedRetention) {
+			if a.State == model.StatePending || (!a.ResolvedAt.IsZero() && ts.Sub(a.ResolvedAt) > ruletypes.ResolvedRetention) {
 				delete(r.Active, fp)
 			}
-			if a.State != ruletypes.StateInactive {
-				a.State = ruletypes.StateInactive
+			if a.State != model.StateInactive {
+				a.State = model.StateInactive
 				a.ResolvedAt = ts
-				itemsToAdd = append(itemsToAdd, rulestatehistorytypes.RuleStateHistory{
+				itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
 					RuleID:       r.ID(),
 					RuleName:     r.Name(),
-					State:        ruletypes.StateInactive,
+					State:        model.StateInactive,
 					StateChanged: true,
 					UnixMilli:    ts.UnixMilli(),
-					Labels:       rulestatehistorytypes.LabelsString(labelsJSON),
-					Fingerprint:  a.QueryResultLabels.Hash(),
+					Labels:       model.LabelsString(labelsJSON),
+					Fingerprint:  a.QueryResultLables.Hash(),
 					Value:        a.Value,
 				})
 			}
 			continue
 		}
 
-		if a.State == ruletypes.StatePending && ts.Sub(a.ActiveAt) >= r.HoldDuration().Duration() {
-			a.State = ruletypes.StateFiring
+		if a.State == model.StatePending && ts.Sub(a.ActiveAt) >= r.HoldDuration().Duration() {
+			a.State = model.StateFiring
 			a.FiredAt = ts
-			state := ruletypes.StateFiring
+			state := model.StateFiring
 			if a.Missing {
-				state = ruletypes.StateNoData
+				state = model.StateNoData
 			}
-			itemsToAdd = append(itemsToAdd, rulestatehistorytypes.RuleStateHistory{
+			itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
 				RuleID:       r.ID(),
 				RuleName:     r.Name(),
 				State:        state,
 				StateChanged: true,
 				UnixMilli:    ts.UnixMilli(),
-				Labels:       rulestatehistorytypes.LabelsString(labelsJSON),
-				Fingerprint:  a.QueryResultLabels.Hash(),
+				Labels:       model.LabelsString(labelsJSON),
+				Fingerprint:  a.QueryResultLables.Hash(),
 				Value:        a.Value,
 			})
 		}
 
 		// We need to change firing alert to recovering if the returned sample meets recovery threshold
-		changeFiringToRecovering := a.State == ruletypes.StateFiring && a.IsRecovering
+		changeFiringToRecovering := a.State == model.StateFiring && a.IsRecovering
 		// We need to change recovering alerts to firing if the returned sample meets target threshold
-		changeRecoveringToFiring := a.State == ruletypes.StateRecovering && !a.IsRecovering && !a.Missing
+		changeRecoveringToFiring := a.State == model.StateRecovering && !a.IsRecovering && !a.Missing
 		// in any of the above case we need to update the status of alert
 		if changeFiringToRecovering || changeRecoveringToFiring {
-			state := ruletypes.StateRecovering
+			state := model.StateRecovering
 			if changeRecoveringToFiring {
-				state = ruletypes.StateFiring
+				state = model.StateFiring
 			}
 			a.State = state
-			r.logger.DebugContext(ctx, "converting alert state", slog.String("rule.id", r.ID()), slog.Any("alert.state", state))
-			itemsToAdd = append(itemsToAdd, rulestatehistorytypes.RuleStateHistory{
+			r.logger.DebugContext(ctx, "converting alert state", "name", r.Name(), "state", state)
+			itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
 				RuleID:       r.ID(),
 				RuleName:     r.Name(),
 				State:        state,
 				StateChanged: true,
 				UnixMilli:    ts.UnixMilli(),
-				Labels:       rulestatehistorytypes.LabelsString(labelsJSON),
-				Fingerprint:  a.QueryResultLabels.Hash(),
+				Labels:       model.LabelsString(labelsJSON),
+				Fingerprint:  a.QueryResultLables.Hash(),
 				Value:        a.Value,
 			})
 		}

ee/query-service/rules/anomaly_test.go

@@ -2,19 +2,21 @@ package rules
 
 import (
 	"context"
+	"log/slog"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/SigNoz/signoz/ee/query-service/anomaly"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/query-service/app/clickhouseReader"
+	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
 	"github.com/SigNoz/signoz/pkg/types/ruletypes"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
-
-	"github.com/SigNoz/signoz/ee/anomaly"
 )
 
 // mockAnomalyProvider is a mock implementation of anomaly.Provider for testing.
@@ -22,13 +24,13 @@ import (
 // time periods (current, past period, current season, past season, past 2 seasons,
 // past 3 seasons), making it cumbersome to create mock data.
 type mockAnomalyProvider struct {
-	responses []*anomaly.AnomaliesResponse
+	responses []*anomaly.GetAnomaliesResponse
 	callCount int
 }
 
-func (m *mockAnomalyProvider) GetAnomalies(ctx context.Context, orgID valuer.UUID, req *anomaly.AnomaliesRequest) (*anomaly.AnomaliesResponse, error) {
+func (m *mockAnomalyProvider) GetAnomalies(ctx context.Context, orgID valuer.UUID, req *anomaly.GetAnomaliesRequest) (*anomaly.GetAnomaliesResponse, error) {
 	if m.callCount >= len(m.responses) {
-		return &anomaly.AnomaliesResponse{Results: []*qbtypes.TimeSeriesData{}}, nil
+		return &anomaly.GetAnomaliesResponse{Results: []*v3.Result{}}, nil
 	}
 	resp := m.responses[m.callCount]
 	m.callCount++
@@ -47,46 +49,45 @@ func TestAnomalyRule_NoData_AlertOnAbsent(t *testing.T) {
 	postableRule := ruletypes.PostableRule{
 		AlertName: "Test anomaly no data",
 		AlertType: ruletypes.AlertTypeMetric,
-		RuleType:  ruletypes.RuleTypeAnomaly,
+		RuleType:  RuleTypeAnomaly,
 		Evaluation: &ruletypes.EvaluationEnvelope{Kind: ruletypes.RollingEvaluation, Spec: ruletypes.RollingWindow{
 			EvalWindow: evalWindow,
 			Frequency:  valuer.MustParseTextDuration("1m"),
 		}},
 		RuleCondition: &ruletypes.RuleCondition{
-			CompareOperator: ruletypes.ValueIsAbove,
-			MatchType:       ruletypes.AtleastOnce,
-			Target:          &target,
-			CompositeQuery: &ruletypes.AlertCompositeQuery{
-				QueryType: ruletypes.QueryTypeBuilder,
-				Queries: []qbtypes.QueryEnvelope{{
-					Type: qbtypes.QueryTypeBuilder,
-					Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
-						Name:   "A",
-						Signal: telemetrytypes.SignalMetrics,
+			CompareOp: ruletypes.ValueIsAbove,
+			MatchType: ruletypes.AtleastOnce,
+			Target:    &target,
+			CompositeQuery: &v3.CompositeQuery{
+				QueryType: v3.QueryTypeBuilder,
+				BuilderQueries: map[string]*v3.BuilderQuery{
+					"A": {
+						QueryName:   "A",
+						Expression:  "A",
+						DataSource:  v3.DataSourceMetrics,
+						Temporality: v3.Unspecified,
 					},
-				}},
+				},
 			},
 			SelectedQuery: "A",
 			Seasonality:   "daily",
 			Thresholds: &ruletypes.RuleThresholdData{
 				Kind: ruletypes.BasicThresholdKind,
 				Spec: ruletypes.BasicRuleThresholds{{
-					Name:            "Test anomaly no data",
-					TargetValue:     &target,
-					MatchType:       ruletypes.AtleastOnce,
-					CompareOperator: ruletypes.ValueIsAbove,
+					Name:        "Test anomaly no data",
+					TargetValue: &target,
+					MatchType:   ruletypes.AtleastOnce,
+					CompareOp:   ruletypes.ValueIsAbove,
 				}},
 			},
 		},
 	}
 
-	responseNoData := &anomaly.AnomaliesResponse{
-		Results: []*qbtypes.TimeSeriesData{
+	responseNoData := &anomaly.GetAnomaliesResponse{
+		Results: []*v3.Result{
 			{
-				QueryName: "A",
-				Aggregations: []*qbtypes.AggregationBucket{{
-					AnomalyScores: []*qbtypes.TimeSeries{},
-				}},
+				QueryName:     "A",
+				AnomalyScores: []*v3.Series{},
 			},
 		},
 	}
@@ -114,17 +115,23 @@ func TestAnomalyRule_NoData_AlertOnAbsent(t *testing.T) {
 		t.Run(c.description, func(t *testing.T) {
 			postableRule.RuleCondition.AlertOnAbsent = c.alertOnAbsent
 
+			telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, nil)
+			options := clickhouseReader.NewOptions("primaryNamespace")
+			reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, nil, "", time.Second, nil, nil, options)
+
 			rule, err := NewAnomalyRule(
 				"test-anomaly-rule",
 				valuer.GenerateUUID(),
 				&postableRule,
+				reader,
 				nil,
 				logger,
+				nil,
 			)
 			require.NoError(t, err)
 
 			rule.provider = &mockAnomalyProvider{
-				responses: []*anomaly.AnomaliesResponse{responseNoData},
+				responses: []*anomaly.GetAnomaliesResponse{responseNoData},
 			}
 
 			alertsFound, err := rule.Eval(context.Background(), evalTime)
@@ -149,47 +156,46 @@ func TestAnomalyRule_NoData_AbsentFor(t *testing.T) {
 	postableRule := ruletypes.PostableRule{
 		AlertName: "Test anomaly no data with AbsentFor",
 		AlertType: ruletypes.AlertTypeMetric,
-		RuleType:  ruletypes.RuleTypeAnomaly,
+		RuleType:  RuleTypeAnomaly,
 		Evaluation: &ruletypes.EvaluationEnvelope{Kind: ruletypes.RollingEvaluation, Spec: ruletypes.RollingWindow{
 			EvalWindow: evalWindow,
 			Frequency:  valuer.MustParseTextDuration("1m"),
 		}},
 		RuleCondition: &ruletypes.RuleCondition{
-			CompareOperator: ruletypes.ValueIsAbove,
-			MatchType:       ruletypes.AtleastOnce,
-			AlertOnAbsent:   true,
-			Target:          &target,
-			CompositeQuery: &ruletypes.AlertCompositeQuery{
-				QueryType: ruletypes.QueryTypeBuilder,
-				Queries: []qbtypes.QueryEnvelope{{
-					Type: qbtypes.QueryTypeBuilder,
-					Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
-						Name:   "A",
-						Signal: telemetrytypes.SignalMetrics,
+			CompareOp:     ruletypes.ValueIsAbove,
+			MatchType:     ruletypes.AtleastOnce,
+			AlertOnAbsent: true,
+			Target:        &target,
+			CompositeQuery: &v3.CompositeQuery{
+				QueryType: v3.QueryTypeBuilder,
+				BuilderQueries: map[string]*v3.BuilderQuery{
+					"A": {
+						QueryName:   "A",
+						Expression:  "A",
+						DataSource:  v3.DataSourceMetrics,
+						Temporality: v3.Unspecified,
 					},
-				}},
+				},
 			},
 			SelectedQuery: "A",
 			Seasonality:   "daily",
 			Thresholds: &ruletypes.RuleThresholdData{
 				Kind: ruletypes.BasicThresholdKind,
 				Spec: ruletypes.BasicRuleThresholds{{
-					Name:            "Test anomaly no data with AbsentFor",
-					TargetValue:     &target,
-					MatchType:       ruletypes.AtleastOnce,
-					CompareOperator: ruletypes.ValueIsAbove,
+					Name:        "Test anomaly no data with AbsentFor",
+					TargetValue: &target,
+					MatchType:   ruletypes.AtleastOnce,
+					CompareOp:   ruletypes.ValueIsAbove,
 				}},
 			},
 		},
 	}
 
-	responseNoData := &anomaly.AnomaliesResponse{
-		Results: []*qbtypes.TimeSeriesData{
+	responseNoData := &anomaly.GetAnomaliesResponse{
+		Results: []*v3.Result{
 			{
-				QueryName: "A",
-				Aggregations: []*qbtypes.AggregationBucket{{
-					AnomalyScores: []*qbtypes.TimeSeries{},
-				}},
+				QueryName:     "A",
+				AnomalyScores: []*v3.Series{},
 			},
 		},
 	}
@@ -223,35 +229,32 @@ func TestAnomalyRule_NoData_AbsentFor(t *testing.T) {
 			t1 := baseTime.Add(5 * time.Minute)
 			t2 := t1.Add(c.timeBetweenEvals)
 
-			responseWithData := &anomaly.AnomaliesResponse{
-				Results: []*qbtypes.TimeSeriesData{
+			responseWithData := &anomaly.GetAnomaliesResponse{
+				Results: []*v3.Result{
 					{
 						QueryName: "A",
-						Aggregations: []*qbtypes.AggregationBucket{{
-							AnomalyScores: []*qbtypes.TimeSeries{
-								{
-									Labels: []*qbtypes.Label{
-										{
-											Key:   telemetrytypes.TelemetryFieldKey{Name: "Test"},
-											Value: "labels",
-										},
-									},
-									Values: []*qbtypes.TimeSeriesValue{
-										{Timestamp: baseTime.UnixMilli(), Value: 1.0},
-										{Timestamp: baseTime.Add(time.Minute).UnixMilli(), Value: 1.5},
-									},
+						AnomalyScores: []*v3.Series{
+							{
+								Labels: map[string]string{"test": "label"},
+								Points: []v3.Point{
+									{Timestamp: baseTime.UnixMilli(), Value: 1.0},
+									{Timestamp: baseTime.Add(time.Minute).UnixMilli(), Value: 1.5},
 								},
 							},
-						}},
+						},
 					},
 				},
 			}
 
-			rule, err := NewAnomalyRule("test-anomaly-rule", valuer.GenerateUUID(), &postableRule, nil, logger)
+			telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, nil)
+			options := clickhouseReader.NewOptions("primaryNamespace")
+			reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, nil, "", time.Second, nil, nil, options)
+
+			rule, err := NewAnomalyRule("test-anomaly-rule", valuer.GenerateUUID(), &postableRule, reader, nil, logger, nil)
 			require.NoError(t, err)
 
 			rule.provider = &mockAnomalyProvider{
-				responses: []*anomaly.AnomaliesResponse{responseWithData, responseNoData},
+				responses: []*anomaly.GetAnomaliesResponse{responseWithData, responseNoData},
 			}
 
 			alertsFound1, err := rule.Eval(context.Background(), t1)

ee/query-service/rules/manager.go

@@ -11,7 +11,9 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/SigNoz/signoz/pkg/errors"
+	basemodel "github.com/SigNoz/signoz/pkg/query-service/model"
 	baserules "github.com/SigNoz/signoz/pkg/query-service/rules"
+	"github.com/SigNoz/signoz/pkg/query-service/utils/labels"
 	"github.com/SigNoz/signoz/pkg/types/ruletypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
@@ -21,7 +23,7 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 	rules := make([]baserules.Rule, 0)
 	var task baserules.Task
 
-	ruleID := baserules.RuleIDFromTaskName(opts.TaskName)
+	ruleId := baserules.RuleIdFromTaskName(opts.TaskName)
 	evaluation, err := opts.Rule.Evaluation.GetEvaluation()
 	if err != nil {
 		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "evaluation is invalid: %v", err)
@@ -30,9 +32,10 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 	if opts.Rule.RuleType == ruletypes.RuleTypeThreshold {
 		// create a threshold rule
 		tr, err := baserules.NewThresholdRule(
-			ruleID,
+			ruleId,
 			opts.OrgID,
 			opts.Rule,
+			opts.Reader,
 			opts.Querier,
 			opts.Logger,
 			baserules.WithEvalDelay(opts.ManagerOpts.EvalDelay),
@@ -55,10 +58,11 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 
 		// create promql rule
 		pr, err := baserules.NewPromRule(
-			ruleID,
+			ruleId,
 			opts.OrgID,
 			opts.Rule,
 			opts.Logger,
+			opts.Reader,
 			opts.ManagerOpts.Prometheus,
 			baserules.WithSQLStore(opts.SQLStore),
 			baserules.WithQueryParser(opts.ManagerOpts.QueryParser),
@@ -78,11 +82,13 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 	} else if opts.Rule.RuleType == ruletypes.RuleTypeAnomaly {
 		// create anomaly rule
 		ar, err := NewAnomalyRule(
-			ruleID,
+			ruleId,
 			opts.OrgID,
 			opts.Rule,
+			opts.Reader,
 			opts.Querier,
 			opts.Logger,
+			opts.Cache,
 			baserules.WithEvalDelay(opts.ManagerOpts.EvalDelay),
 			baserules.WithSQLStore(opts.SQLStore),
 			baserules.WithQueryParser(opts.ManagerOpts.QueryParser),
@@ -99,7 +105,7 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 		task = newTask(baserules.TaskTypeCh, opts.TaskName, evaluation.GetFrequency().Duration(), rules, opts.ManagerOpts, opts.NotifyFunc, opts.MaintenanceStore, opts.OrgID)
 
 	} else {
-		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported rule type %s. Supported types: %s, %s", opts.Rule.RuleType, ruletypes.RuleTypeProm, ruletypes.RuleTypeThreshold)
+		return nil, fmt.Errorf("unsupported rule type %s. Supported types: %s, %s", opts.Rule.RuleType, ruletypes.RuleTypeProm, ruletypes.RuleTypeThreshold)
 	}
 
 	return task, nil
@@ -107,12 +113,12 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 
 // TestNotification prepares a dummy rule for given rule parameters and
 // sends a test notification. returns alert count and error (if any)
-func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
+func TestNotification(opts baserules.PrepareTestRuleOptions) (int, *basemodel.ApiError) {
 
 	ctx := context.Background()
 
 	if opts.Rule == nil {
-		return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "rule is required")
+		return 0, basemodel.BadRequest(fmt.Errorf("rule is required"))
 	}
 
 	parsedRule := opts.Rule
@@ -132,14 +138,15 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 	if parsedRule.RuleType == ruletypes.RuleTypeThreshold {
 
 		// add special labels for test alerts
-		parsedRule.Labels[ruletypes.RuleSourceLabel] = ""
-		parsedRule.Labels[ruletypes.AlertRuleIDLabel] = ""
+		parsedRule.Labels[labels.RuleSourceLabel] = ""
+		parsedRule.Labels[labels.AlertRuleIdLabel] = ""
 
 		// create a threshold rule
 		rule, err = baserules.NewThresholdRule(
 			alertname,
 			opts.OrgID,
 			parsedRule,
+			opts.Reader,
 			opts.Querier,
 			opts.Logger,
 			baserules.WithSendAlways(),
@@ -151,7 +158,7 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 
 		if err != nil {
 			slog.Error("failed to prepare a new threshold rule for test", "name", alertname, errors.Attr(err))
-			return 0, err
+			return 0, basemodel.BadRequest(err)
 		}
 
 	} else if parsedRule.RuleType == ruletypes.RuleTypeProm {
@@ -162,6 +169,7 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 			opts.OrgID,
 			parsedRule,
 			opts.Logger,
+			opts.Reader,
 			opts.ManagerOpts.Prometheus,
 			baserules.WithSendAlways(),
 			baserules.WithSendUnmatched(),
@@ -172,7 +180,7 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 
 		if err != nil {
 			slog.Error("failed to prepare a new promql rule for test", "name", alertname, errors.Attr(err))
-			return 0, err
+			return 0, basemodel.BadRequest(err)
 		}
 	} else if parsedRule.RuleType == ruletypes.RuleTypeAnomaly {
 		// create anomaly rule
@@ -180,8 +188,10 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 			alertname,
 			opts.OrgID,
 			parsedRule,
+			opts.Reader,
 			opts.Querier,
 			opts.Logger,
+			opts.Cache,
 			baserules.WithSendAlways(),
 			baserules.WithSendUnmatched(),
 			baserules.WithSQLStore(opts.SQLStore),
@@ -190,10 +200,10 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 		)
 		if err != nil {
 			slog.Error("failed to prepare a new anomaly rule for test", "name", alertname, errors.Attr(err))
-			return 0, err
+			return 0, basemodel.BadRequest(err)
 		}
 	} else {
-		return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "failed to derive ruletype with given information")
+		return 0, basemodel.BadRequest(fmt.Errorf("failed to derive ruletype with given information"))
 	}
 
 	// set timestamp to current utc time
@@ -202,7 +212,7 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
 	alertsFound, err := rule.Eval(ctx, ts)
 	if err != nil {
 		slog.Error("evaluating rule failed", "rule", rule.Name(), errors.Attr(err))
-		return 0, err
+		return 0, basemodel.InternalError(fmt.Errorf("rule evaluation failed"))
 	}
 	rule.SendAlerts(ctx, ts, 0, time.Minute, opts.NotifyFunc)
 

ee/query-service/rules/manager_test.go

@@ -114,8 +114,11 @@ func TestManager_TestNotification_SendUnmatched_ThresholdRule(t *testing.T) {
 				},
 			})
 
-			count, err := mgr.TestNotification(context.Background(), orgID, string(ruleBytes))
-			require.Nil(t, err)
+			count, apiErr := mgr.TestNotification(context.Background(), orgID, string(ruleBytes))
+			if apiErr != nil {
+				t.Logf("TestNotification error: %v, type: %s", apiErr.Err, apiErr.Typ)
+			}
+			require.Nil(t, apiErr)
 			assert.Equal(t, tc.ExpectAlerts, count)
 
 			if tc.ExpectAlerts > 0 {
@@ -265,8 +268,11 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
 				},
 			})
 
-			count, err := mgr.TestNotification(context.Background(), orgID, string(ruleBytes))
-			require.Nil(t, err)
+			count, apiErr := mgr.TestNotification(context.Background(), orgID, string(ruleBytes))
+			if apiErr != nil {
+				t.Logf("TestNotification error: %v, type: %s", apiErr.Err, apiErr.Typ)
+			}
+			require.Nil(t, apiErr)
 			assert.Equal(t, tc.ExpectAlerts, count)
 
 			if tc.ExpectAlerts > 0 {

frontend/package.json

@@ -68,8 +68,8 @@
     "@signozhq/toggle-group": "0.0.1",
     "@signozhq/tooltip": "0.0.2",
     "@signozhq/ui": "0.0.5",
-    "@tanstack/react-table": "8.21.3",
-    "@tanstack/react-virtual": "3.13.22",
+    "@tanstack/react-table": "8.20.6",
+    "@tanstack/react-virtual": "3.11.2",
     "@uiw/codemirror-theme-copilot": "4.23.11",
     "@uiw/codemirror-theme-github": "4.24.1",
     "@uiw/react-codemirror": "4.23.10",