refactor: updating response for conn status

ee/query-service/app/api/api.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"log/slog"
 	"net/http"
 	"net/http/httputil"
 	"time"
@@ -14,7 +15,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/http/middleware"
 	querierAPI "github.com/SigNoz/signoz/pkg/querier"
 	baseapp "github.com/SigNoz/signoz/pkg/query-service/app"
-	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations"
 	"github.com/SigNoz/signoz/pkg/query-service/app/integrations"
 	"github.com/SigNoz/signoz/pkg/query-service/app/logparsingpipeline"
 	"github.com/SigNoz/signoz/pkg/query-service/interfaces"
@@ -22,6 +22,7 @@ import (
 	rules "github.com/SigNoz/signoz/pkg/query-service/rules"
 	"github.com/SigNoz/signoz/pkg/queryparser"
 	"github.com/SigNoz/signoz/pkg/signoz"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/version"
 	"github.com/gorilla/mux"
 )
@@ -31,13 +32,14 @@ type APIHandlerOptions struct {
 	RulesManager                  *rules.Manager
 	UsageManager                  *usage.Manager
 	IntegrationsController        *integrations.Controller
-	CloudIntegrationsController   *cloudintegrations.Controller
+	CloudIntegrationsRegistry     map[integrationstypes.CloudProviderType]integrationstypes.CloudProvider
 	LogsParsingPipelineController *logparsingpipeline.LogParsingPipelineController
 	Gateway                       *httputil.ReverseProxy
 	GatewayUrl                    string
 	// Querier Influx Interval
 	FluxInterval time.Duration
 	GlobalConfig global.Config
+	Logger       *slog.Logger // this is present in Signoz.Instrumentation but adding for quick access
 }
 
 type APIHandler struct {
@@ -51,7 +53,7 @@ func NewAPIHandler(opts APIHandlerOptions, signoz *signoz.SigNoz) (*APIHandler,
 		Reader:                        opts.DataConnector,
 		RuleManager:                   opts.RulesManager,
 		IntegrationsController:        opts.IntegrationsController,
-		CloudIntegrationsController:   opts.CloudIntegrationsController,
+		CloudIntegrationsRegistry:     opts.CloudIntegrationsRegistry,
 		LogsParsingPipelineController: opts.LogsParsingPipelineController,
 		FluxInterval:                  opts.FluxInterval,
 		AlertmanagerAPI:               alertmanager.NewAPI(signoz.Alertmanager),
@@ -120,14 +122,12 @@ func (ah *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
 }
 
 func (ah *APIHandler) RegisterCloudIntegrationsRoutes(router *mux.Router, am *middleware.AuthZ) {
-
 	ah.APIHandler.RegisterCloudIntegrationsRoutes(router, am)
 
 	router.HandleFunc(
 		"/api/v1/cloud-integrations/{cloudProvider}/accounts/generate-connection-params",
 		am.EditAccess(ah.CloudIntegrationsGenerateConnectionParams),
 	).Methods(http.MethodGet)
-
 }
 
 func (ah *APIHandler) getVersion(w http.ResponseWriter, r *http.Request) {

ee/query-service/app/api/cloudIntegrations.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"strings"
 	"time"
@@ -13,20 +14,14 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/http/render"
 	"github.com/SigNoz/signoz/pkg/modules/user"
-	basemodel "github.com/SigNoz/signoz/pkg/query-service/model"
 	"github.com/SigNoz/signoz/pkg/types"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/gorilla/mux"
-	"go.uber.org/zap"
 )
 
-type CloudIntegrationConnectionParamsResponse struct {
-	IngestionUrl string `json:"ingestion_url,omitempty"`
-	IngestionKey string `json:"ingestion_key,omitempty"`
-	SigNozAPIUrl string `json:"signoz_api_url,omitempty"`
-	SigNozAPIKey string `json:"signoz_api_key,omitempty"`
-}
+// TODO: move this file with other cloud integration related code
 
 func (ah *APIHandler) CloudIntegrationsGenerateConnectionParams(w http.ResponseWriter, r *http.Request) {
 	claims, err := authtypes.ClaimsFromContext(r.Context())
@@ -41,23 +36,21 @@ func (ah *APIHandler) CloudIntegrationsGenerateConnectionParams(w http.ResponseW
 		return
 	}
 
-	cloudProvider := mux.Vars(r)["cloudProvider"]
-	if cloudProvider != "aws" {
-		RespondError(w, basemodel.BadRequest(fmt.Errorf(
-			"cloud provider not supported: %s", cloudProvider,
-		)), nil)
+	cloudProviderString := mux.Vars(r)["cloudProvider"]
+
+	cloudProvider, err := integrationstypes.NewCloudProvider(cloudProviderString)
+	if err != nil {
+		render.Error(w, err)
 		return
 	}
 
-	apiKey, apiErr := ah.getOrCreateCloudIntegrationPAT(r.Context(), claims.OrgID, cloudProvider)
-	if apiErr != nil {
-		RespondError(w, basemodel.WrapApiError(
-			apiErr, "couldn't provision PAT for cloud integration:",
-		), nil)
+	apiKey, err := ah.getOrCreateCloudIntegrationPAT(r.Context(), claims.OrgID, cloudProvider)
+	if err != nil {
+		render.Error(w, err)
 		return
 	}
 
-	result := CloudIntegrationConnectionParamsResponse{
+	result := integrationstypes.GettableCloudIntegrationConnectionParams{
 		SigNozAPIKey: apiKey,
 	}
 
@@ -71,16 +64,17 @@ func (ah *APIHandler) CloudIntegrationsGenerateConnectionParams(w http.ResponseW
 		// Return the API Key (PAT) even if the rest of the params can not be deduced.
 		// Params not returned from here will be requested from the user via form inputs.
 		// This enables gracefully degraded but working experience even for non-cloud deployments.
-		zap.L().Info("ingestion params and signoz api url can not be deduced since no license was found")
-		ah.Respond(w, result)
+		ah.opts.Logger.InfoContext(
+			r.Context(),
+			"ingestion params and signoz api url can not be deduced since no license was found",
+		)
+		render.Success(w, http.StatusOK, result)
 		return
 	}
 
-	signozApiUrl, apiErr := ah.getIngestionUrlAndSigNozAPIUrl(r.Context(), license.Key)
-	if apiErr != nil {
-		RespondError(w, basemodel.WrapApiError(
-			apiErr, "couldn't deduce ingestion url and signoz api url",
-		), nil)
+	signozApiUrl, err := ah.getIngestionUrlAndSigNozAPIUrl(r.Context(), license.Key)
+	if err != nil {
+		render.Error(w, err)
 		return
 	}
 
@@ -89,48 +83,41 @@ func (ah *APIHandler) CloudIntegrationsGenerateConnectionParams(w http.ResponseW
 
 	gatewayUrl := ah.opts.GatewayUrl
 	if len(gatewayUrl) > 0 {
-
-		ingestionKey, apiErr := getOrCreateCloudProviderIngestionKey(
+		ingestionKeyString, err := ah.getOrCreateCloudProviderIngestionKey(
 			r.Context(), gatewayUrl, license.Key, cloudProvider,
 		)
-		if apiErr != nil {
-			RespondError(w, basemodel.WrapApiError(
-				apiErr, "couldn't get or create ingestion key",
-			), nil)
+		if err != nil {
+			render.Error(w, err)
 			return
 		}
 
-		result.IngestionKey = ingestionKey
-
+		result.IngestionKey = ingestionKeyString
 	} else {
-		zap.L().Info("ingestion key can't be deduced since no gateway url has been configured")
+		ah.opts.Logger.InfoContext(
+			r.Context(),
+			"ingestion key can't be deduced since no gateway url has been configured",
+		)
 	}
 
-	ah.Respond(w, result)
+	render.Success(w, http.StatusOK, result)
 }
 
-func (ah *APIHandler) getOrCreateCloudIntegrationPAT(ctx context.Context, orgId string, cloudProvider string) (
-	string, *basemodel.ApiError,
-) {
+func (ah *APIHandler) getOrCreateCloudIntegrationPAT(ctx context.Context, orgId string, cloudProvider valuer.String) (string, error) {
 	integrationPATName := fmt.Sprintf("%s integration", cloudProvider)
 
-	integrationUser, apiErr := ah.getOrCreateCloudIntegrationUser(ctx, orgId, cloudProvider)
-	if apiErr != nil {
-		return "", apiErr
+	integrationUser, err := ah.getOrCreateCloudIntegrationUser(ctx, orgId, cloudProvider)
+	if err != nil {
+		return "", err
 	}
 
 	orgIdUUID, err := valuer.NewUUID(orgId)
 	if err != nil {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't parse orgId: %w", err,
-		))
+		return "", err
 	}
 
 	allPats, err := ah.Signoz.Modules.User.ListAPIKeys(ctx, orgIdUUID)
 	if err != nil {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't list PATs: %w", err,
-		))
+		return "", err
 	}
 	for _, p := range allPats {
 		if p.UserID == integrationUser.ID && p.Name == integrationPATName {
@@ -138,9 +125,10 @@ func (ah *APIHandler) getOrCreateCloudIntegrationPAT(ctx context.Context, orgId
 		}
 	}
 
-	zap.L().Info(
+	ah.opts.Logger.InfoContext(
+		ctx,
 		"no PAT found for cloud integration, creating a new one",
-		zap.String("cloudProvider", cloudProvider),
+		slog.String("cloudProvider", cloudProvider.String()),
 	)
 
 	newPAT, err := types.NewStorableAPIKey(
@@ -150,68 +138,48 @@ func (ah *APIHandler) getOrCreateCloudIntegrationPAT(ctx context.Context, orgId
 		0,
 	)
 	if err != nil {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't create cloud integration PAT: %w", err,
-		))
+		return "", err
 	}
 
 	err = ah.Signoz.Modules.User.CreateAPIKey(ctx, newPAT)
 	if err != nil {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't create cloud integration PAT: %w", err,
-		))
+		return "", err
 	}
 	return newPAT.Token, nil
 }
 
-func (ah *APIHandler) getOrCreateCloudIntegrationUser(
-	ctx context.Context, orgId string, cloudProvider string,
-) (*types.User, *basemodel.ApiError) {
-	cloudIntegrationUserName := fmt.Sprintf("%s-integration", cloudProvider)
+// TODO: move this function out of handler and use proper module structure
+func (ah *APIHandler) getOrCreateCloudIntegrationUser(ctx context.Context, orgId string, cloudProvider valuer.String) (*types.User, error) {
+	cloudIntegrationUserName := fmt.Sprintf("%s-integration", cloudProvider.String())
 	email := valuer.MustNewEmail(fmt.Sprintf("%s@signoz.io", cloudIntegrationUserName))
 
 	cloudIntegrationUser, err := types.NewUser(cloudIntegrationUserName, email, types.RoleViewer, valuer.MustNewUUID(orgId))
 	if err != nil {
-		return nil, basemodel.InternalError(fmt.Errorf("couldn't create cloud integration user: %w", err))
+		return nil, err
 	}
 
 	password := types.MustGenerateFactorPassword(cloudIntegrationUser.ID.StringValue())
 
 	cloudIntegrationUser, err = ah.Signoz.Modules.User.GetOrCreateUser(ctx, cloudIntegrationUser, user.WithFactorPassword(password))
 	if err != nil {
-		return nil, basemodel.InternalError(fmt.Errorf("couldn't look for integration user: %w", err))
+		return nil, err
 	}
 
 	return cloudIntegrationUser, nil
 }
 
-func (ah *APIHandler) getIngestionUrlAndSigNozAPIUrl(ctx context.Context, licenseKey string) (
-	string, *basemodel.ApiError,
-) {
-	// TODO: remove this struct from here
-	type deploymentResponse struct {
-		Name        string `json:"name"`
-		ClusterInfo struct {
-			Region struct {
-				DNS string `json:"dns"`
-			} `json:"region"`
-		} `json:"cluster"`
-	}
-
+// TODO: move this function out of handler and use proper module structure
+func (ah *APIHandler) getIngestionUrlAndSigNozAPIUrl(ctx context.Context, licenseKey string) (string, error) {
 	respBytes, err := ah.Signoz.Zeus.GetDeployment(ctx, licenseKey)
 	if err != nil {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't query for deployment info: error: %w", err,
-		))
+		return "", errors.WrapInternalf(err, errors.CodeInternal, "couldn't query for deployment info: error")
 	}
 
-	resp := new(deploymentResponse)
+	resp := new(integrationstypes.GettableDeployment)
 
 	err = json.Unmarshal(respBytes, resp)
 	if err != nil {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't unmarshal deployment info response: error: %w", err,
-		))
+		return "", errors.WrapInternalf(err, errors.CodeInternal, "couldn't unmarshal deployment info response")
 	}
 
 	regionDns := resp.ClusterInfo.Region.DNS
@@ -219,9 +187,11 @@ func (ah *APIHandler) getIngestionUrlAndSigNozAPIUrl(ctx context.Context, licens
 
 	if len(regionDns) < 1 || len(deploymentName) < 1 {
 		// Fail early if actual response structure and expectation here ever diverge
-		return "", basemodel.InternalError(fmt.Errorf(
+		return "", errors.WrapInternalf(
+			err,
+			errors.CodeInternal,
 			"deployment info response not in expected shape. couldn't determine region dns and deployment name",
-		))
+		)
 	}
 
 	signozApiUrl := fmt.Sprintf("https://%s.%s", deploymentName, regionDns)
@@ -229,102 +199,85 @@ func (ah *APIHandler) getIngestionUrlAndSigNozAPIUrl(ctx context.Context, licens
 	return signozApiUrl, nil
 }
 
-type ingestionKey struct {
-	Name  string `json:"name"`
-	Value string `json:"value"`
-	// other attributes from gateway response not included here since they are not being used.
-}
-
-type ingestionKeysSearchResponse struct {
-	Status string         `json:"status"`
-	Data   []ingestionKey `json:"data"`
-	Error  string         `json:"error"`
-}
-
-type createIngestionKeyResponse struct {
-	Status string       `json:"status"`
-	Data   ingestionKey `json:"data"`
-	Error  string       `json:"error"`
-}
-
-func getOrCreateCloudProviderIngestionKey(
-	ctx context.Context, gatewayUrl string, licenseKey string, cloudProvider string,
-) (string, *basemodel.ApiError) {
+func (ah *APIHandler) getOrCreateCloudProviderIngestionKey(
+	ctx context.Context, gatewayUrl string, licenseKey string, cloudProvider valuer.String,
+) (string, error) {
 	cloudProviderKeyName := fmt.Sprintf("%s-integration", cloudProvider)
 
 	// see if the key already exists
-	searchResult, apiErr := requestGateway[ingestionKeysSearchResponse](
+	searchResult, err := requestGateway[integrationstypes.GettableIngestionKeysSearch](
 		ctx,
 		gatewayUrl,
 		licenseKey,
 		fmt.Sprintf("/v1/workspaces/me/keys/search?name=%s", cloudProviderKeyName),
 		nil,
+		ah.opts.Logger,
 	)
-
-	if apiErr != nil {
-		return "", basemodel.WrapApiError(
-			apiErr, "couldn't search for cloudprovider ingestion key",
-		)
+	if err != nil {
+		return "", err
 	}
 
 	if searchResult.Status != "success" {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't search for cloudprovider ingestion key: status: %s, error: %s",
+		return "", errors.NewInternalf(
+			errors.CodeInternal,
+			"couldn't search for cloud provider ingestion key: status: %s, error: %s",
 			searchResult.Status, searchResult.Error,
-		))
-	}
-
-	for _, k := range searchResult.Data {
-		if k.Name == cloudProviderKeyName {
-			if len(k.Value) < 1 {
-				// Fail early if actual response structure and expectation here ever diverge
-				return "", basemodel.InternalError(fmt.Errorf(
-					"ingestion keys search response not as expected",
-				))
-			}
-
-			return k.Value, nil
-		}
-	}
-
-	zap.L().Info(
-		"no existing ingestion key found for cloud integration, creating a new one",
-		zap.String("cloudProvider", cloudProvider),
-	)
-	createKeyResult, apiErr := requestGateway[createIngestionKeyResponse](
-		ctx, gatewayUrl, licenseKey, "/v1/workspaces/me/keys",
-		map[string]any{
-			"name": cloudProviderKeyName,
-			"tags": []string{"integration", cloudProvider},
-		},
-	)
-	if apiErr != nil {
-		return "", basemodel.WrapApiError(
-			apiErr, "couldn't create cloudprovider ingestion key",
 		)
 	}
 
+	for _, k := range searchResult.Data {
+		if k.Name != cloudProviderKeyName {
+			continue
+		}
+
+		if len(k.Value) < 1 {
+			// Fail early if actual response structure and expectation here ever diverge
+			return "", errors.NewInternalf(errors.CodeInternal, "ingestion keys search response not as expected")
+		}
+
+		return k.Value, nil
+	}
+
+	ah.opts.Logger.InfoContext(
+		ctx,
+		"no existing ingestion key found for cloud integration, creating a new one",
+		slog.String("cloudProvider", cloudProvider.String()),
+	)
+
+	createKeyResult, err := requestGateway[integrationstypes.GettableCreateIngestionKey](
+		ctx, gatewayUrl, licenseKey, "/v1/workspaces/me/keys",
+		map[string]any{
+			"name": cloudProviderKeyName,
+			"tags": []string{"integration", cloudProvider.String()},
+		},
+		ah.opts.Logger,
+	)
+	if err != nil {
+		return "", err
+	}
+
 	if createKeyResult.Status != "success" {
-		return "", basemodel.InternalError(fmt.Errorf(
-			"couldn't create cloudprovider ingestion key: status: %s, error: %s",
+		return "", errors.NewInternalf(
+			errors.CodeInternal,
+			"couldn't create cloud provider ingestion key: status: %s, error: %s",
 			createKeyResult.Status, createKeyResult.Error,
-		))
+		)
 	}
 
-	ingestionKey := createKeyResult.Data.Value
-	if len(ingestionKey) < 1 {
+	ingestionKeyString := createKeyResult.Data.Value
+	if len(ingestionKeyString) < 1 {
 		// Fail early if actual response structure and expectation here ever diverge
-		return "", basemodel.InternalError(fmt.Errorf(
+		return "", errors.NewInternalf(errors.CodeInternal,
 			"ingestion key creation response not as expected",
-		))
+		)
 	}
 
-	return ingestionKey, nil
+	return ingestionKeyString, nil
 }
 
 func requestGateway[ResponseType any](
-	ctx context.Context, gatewayUrl string, licenseKey string, path string, payload any,
-) (*ResponseType, *basemodel.ApiError) {
+	ctx context.Context, gatewayUrl, licenseKey, path string, payload any, logger *slog.Logger,
+) (*ResponseType, error) {
 
 	baseUrl := strings.TrimSuffix(gatewayUrl, "/")
 	reqUrl := fmt.Sprintf("%s%s", baseUrl, path)
@@ -335,13 +288,12 @@ func requestGateway[ResponseType any](
 		"X-Consumer-Groups":      "ns:default",
 	}
 
-	return requestAndParseResponse[ResponseType](ctx, reqUrl, headers, payload)
+	return requestAndParseResponse[ResponseType](ctx, reqUrl, headers, payload, logger)
 }
 
 func requestAndParseResponse[ResponseType any](
-	ctx context.Context, url string, headers map[string]string, payload any,
-) (*ResponseType, *basemodel.ApiError) {
-
+	ctx context.Context, url string, headers map[string]string, payload any, logger *slog.Logger,
+) (*ResponseType, error) {
 	reqMethod := http.MethodGet
 	var reqBody io.Reader
 	if payload != nil {
@@ -349,18 +301,14 @@ func requestAndParseResponse[ResponseType any](
 
 		bodyJson, err := json.Marshal(payload)
 		if err != nil {
-			return nil, basemodel.InternalError(fmt.Errorf(
-				"couldn't serialize request payload to JSON: %w", err,
-			))
+			return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't marshal payload")
 		}
-		reqBody = bytes.NewBuffer([]byte(bodyJson))
+		reqBody = bytes.NewBuffer(bodyJson)
 	}
 
 	req, err := http.NewRequestWithContext(ctx, reqMethod, url, reqBody)
 	if err != nil {
-		return nil, basemodel.InternalError(fmt.Errorf(
-			"couldn't prepare request: %w", err,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't create req")
 	}
 
 	for k, v := range headers {
@@ -373,23 +321,26 @@ func requestAndParseResponse[ResponseType any](
 
 	response, err := client.Do(req)
 	if err != nil {
-		return nil, basemodel.InternalError(fmt.Errorf("couldn't make request: %w", err))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't make req")
 	}
 
-	defer response.Body.Close()
+	defer func() {
+		err = response.Body.Close()
+		if err != nil {
+			logger.ErrorContext(ctx, "couldn't close response body", "error", err)
+		}
+	}()
 
 	respBody, err := io.ReadAll(response.Body)
 	if err != nil {
-		return nil, basemodel.InternalError(fmt.Errorf("couldn't read response: %w", err))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't read response body")
 	}
 
 	var resp ResponseType
 
 	err = json.Unmarshal(respBody, &resp)
 	if err != nil {
-		return nil, basemodel.InternalError(fmt.Errorf(
-			"couldn't unmarshal gateway response into %T", resp,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't unmarshal response body")
 	}
 
 	return &resp, nil

ee/query-service/app/server.go

@@ -127,12 +127,7 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 		)
 	}
 
-	cloudIntegrationsController, err := cloudintegrations.NewController(signoz.SQLStore)
-	if err != nil {
-		return nil, fmt.Errorf(
-			"couldn't create cloud provider integrations controller: %w", err,
-		)
-	}
+	cloudIntegrationsRegistry := cloudintegrations.NewCloudProviderRegistry(signoz.Instrumentation.Logger(), signoz.SQLStore, signoz.Querier)
 
 	// ingestion pipelines manager
 	logParsingPipelineController, err := logparsingpipeline.NewLogParsingPipelinesController(
@@ -167,12 +162,13 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 		RulesManager:                  rm,
 		UsageManager:                  usageManager,
 		IntegrationsController:        integrationsController,
-		CloudIntegrationsController:   cloudIntegrationsController,
+		CloudIntegrationsRegistry:     cloudIntegrationsRegistry,
 		LogsParsingPipelineController: logParsingPipelineController,
 		FluxInterval:                  config.Querier.FluxInterval,
 		Gateway:                       gatewayProxy,
 		GatewayUrl:                    config.Gateway.URL.String(),
 		GlobalConfig:                  config.Global,
+		Logger:                        signoz.Instrumentation.Logger(),
 	}
 
 	apiHandler, err := api.NewAPIHandler(apiOpts, signoz)

pkg/http/middleware/recovery.go

@@ -0,0 +1,44 @@
+package middleware
+
+import (
+	"log/slog"
+	"net/http"
+	"runtime/debug"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/http/render"
+)
+
+// Recovery is a middleware that recovers from panics, logs the panic,
+// and returns a 500 Internal Server Error.
+type Recovery struct {
+	logger *slog.Logger
+}
+
+// NewRecovery creates a new Recovery middleware.
+func NewRecovery(logger *slog.Logger) Wrapper {
+	return &Recovery{
+		logger: logger.With("pkg", "http-middleware-recovery"),
+	}
+}
+
+// Wrap is the middleware handler.
+func (m *Recovery) Wrap(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer func() {
+			if err := recover(); err != nil {
+				m.logger.ErrorContext(
+					r.Context(),
+					"panic recovered",
+					"err", err, "stack", string(debug.Stack()),
+				)
+
+				render.Error(w, errors.NewInternalf(
+					errors.CodeInternal, "internal server error",
+				))
+			}
+		}()
+
+		next.ServeHTTP(w, r)
+	})
+}

pkg/modules/user/impluser/handler.go

@@ -13,6 +13,7 @@ import (
 	root "github.com/SigNoz/signoz/pkg/modules/user"
 	"github.com/SigNoz/signoz/pkg/types"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/gorilla/mux"
 )
@@ -462,7 +463,7 @@ func (h *handler) UpdateAPIKey(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if slices.Contains(types.AllIntegrationUserEmails, types.IntegrationUserEmail(createdByUser.Email.String())) {
+	if slices.Contains(integrationstypes.IntegrationUserEmails, createdByUser.Email) {
 		render.Error(w, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "API Keys for integration users cannot be revoked"))
 		return
 	}
@@ -507,7 +508,7 @@ func (h *handler) RevokeAPIKey(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if slices.Contains(types.AllIntegrationUserEmails, types.IntegrationUserEmail(createdByUser.Email.String())) {
+	if slices.Contains(integrationstypes.IntegrationUserEmails, createdByUser.Email) {
 		render.Error(w, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "API Keys for integration users cannot be revoked"))
 		return
 	}

pkg/modules/user/impluser/module.go

@@ -19,6 +19,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/types"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
 	"github.com/SigNoz/signoz/pkg/types/emailtypes"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/types/roletypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/dustin/go-humanize"
@@ -171,7 +172,7 @@ func (m *Module) DeleteInvite(ctx context.Context, orgID string, id valuer.UUID)
 func (module *Module) CreateUser(ctx context.Context, input *types.User, opts ...root.CreateUserOption) error {
 	createUserOpts := root.NewCreateUserOptions(opts...)
 
-	// since assign is idempotant multiple calls to assign won't cause issues in case of retries.
+	// since assign is idempotent multiple calls to assign won't cause issues in case of retries.
 	err := module.granter.Grant(ctx, input.OrgID, roletypes.MustGetSigNozManagedRoleFromExistingRole(input.Role), authtypes.MustNewSubject(authtypes.TypeableUser, input.ID.StringValue(), input.OrgID, nil))
 	if err != nil {
 		return err
@@ -286,7 +287,7 @@ func (module *Module) DeleteUser(ctx context.Context, orgID valuer.UUID, id stri
 		return err
 	}
 
-	if slices.Contains(types.AllIntegrationUserEmails, types.IntegrationUserEmail(user.Email.String())) {
+	if slices.Contains(integrationstypes.IntegrationUserEmails, user.Email) {
 		return errors.New(errors.TypeForbidden, errors.CodeForbidden, "integration user cannot be deleted")
 	}
 
@@ -300,7 +301,7 @@ func (module *Module) DeleteUser(ctx context.Context, orgID valuer.UUID, id stri
 		return errors.New(errors.TypeForbidden, errors.CodeForbidden, "cannot delete the last admin")
 	}
 
-	// since revoke is idempotant multiple calls to revoke won't cause issues in case of retries
+	// since revoke is idempotent multiple calls to revoke won't cause issues in case of retries
 	err = module.granter.Revoke(ctx, orgID, roletypes.MustGetSigNozManagedRoleFromExistingRole(user.Role), authtypes.MustNewSubject(authtypes.TypeableUser, id, orgID, nil))
 	if err != nil {
 		return err

pkg/query-service/app/cloudintegrations/implawsprovider/awsprovider.go

@@ -0,0 +1,796 @@
+package implawsprovider
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/url"
+	"slices"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/querier"
+	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/services"
+	integrationstore "github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/store"
+	"github.com/SigNoz/signoz/pkg/query-service/model"
+	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
+	"github.com/SigNoz/signoz/pkg/types/metrictypes"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"golang.org/x/exp/maps"
+)
+
+var (
+	CodeInvalidAWSRegion  = errors.MustNewCode("invalid_aws_region")
+	CodeDashboardNotFound = errors.MustNewCode("dashboard_not_found")
+)
+
+type awsProvider struct {
+	logger                *slog.Logger
+	querier               querier.Querier
+	accountsRepo          integrationstore.CloudProviderAccountsRepository
+	serviceConfigRepo     integrationstore.ServiceConfigDatabase
+	awsServiceDefinitions *services.AWSServicesProvider
+}
+
+func NewAWSCloudProvider(
+	logger *slog.Logger,
+	accountsRepo integrationstore.CloudProviderAccountsRepository,
+	serviceConfigRepo integrationstore.ServiceConfigDatabase,
+	querier querier.Querier,
+) integrationstypes.CloudProvider {
+	awsServiceDefinitions, err := services.NewAWSCloudProviderServices()
+	if err != nil {
+		panic("failed to initialize AWS service definitions: " + err.Error())
+	}
+
+	return &awsProvider{
+		logger:                logger,
+		querier:               querier,
+		accountsRepo:          accountsRepo,
+		serviceConfigRepo:     serviceConfigRepo,
+		awsServiceDefinitions: awsServiceDefinitions,
+	}
+}
+
+func (a *awsProvider) GetAccountStatus(ctx context.Context, orgID, accountID string) (*integrationstypes.GettableAccountStatus, error) {
+	accountRecord, err := a.accountsRepo.Get(ctx, orgID, a.GetName().String(), accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	return &integrationstypes.GettableAccountStatus{
+		Id:             accountRecord.ID.String(),
+		CloudAccountId: accountRecord.AccountID,
+		Status:         accountRecord.Status(),
+	}, nil
+}
+
+func (a *awsProvider) ListConnectedAccounts(ctx context.Context, orgID string) (*integrationstypes.GettableConnectedAccountsList, error) {
+	accountRecords, err := a.accountsRepo.ListConnected(ctx, orgID, a.GetName().String())
+	if err != nil {
+		return nil, err
+	}
+
+	connectedAccounts := make([]*integrationstypes.Account, 0, len(accountRecords))
+	for _, r := range accountRecords {
+		connectedAccounts = append(connectedAccounts, r.Account(a.GetName()))
+	}
+
+	return &integrationstypes.GettableConnectedAccountsList{
+		Accounts: connectedAccounts,
+	}, nil
+}
+
+func (a *awsProvider) AgentCheckIn(ctx context.Context, req *integrationstypes.PostableAgentCheckInPayload) (any, error) {
+	// agent can't check in unless the account is already created
+	existingAccount, err := a.accountsRepo.Get(ctx, req.OrgID, a.GetName().String(), req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	if existingAccount != nil && existingAccount.AccountID != nil && *existingAccount.AccountID != req.AccountID {
+		return nil, model.BadRequest(fmt.Errorf(
+			"can't check in with new %s account id %s for account %s with existing %s id %s",
+			a.GetName().String(), req.AccountID, existingAccount.ID.StringValue(), a.GetName().String(),
+			*existingAccount.AccountID,
+		))
+	}
+
+	existingAccount, err = a.accountsRepo.GetConnectedCloudAccount(ctx, req.OrgID, a.GetName().String(), req.AccountID)
+	if existingAccount != nil && existingAccount.ID.StringValue() != req.ID {
+		return nil, model.BadRequest(fmt.Errorf(
+			"can't check in to %s account %s with id %s. already connected with id %s",
+			a.GetName().String(), req.AccountID, req.ID, existingAccount.ID.StringValue(),
+		))
+	}
+
+	agentReport := integrationstypes.AgentReport{
+		TimestampMillis: time.Now().UnixMilli(),
+		Data:            req.Data,
+	}
+
+	account, err := a.accountsRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), &req.ID, nil, &req.AccountID, &agentReport, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	agentConfig, err := a.getAWSAgentConfig(ctx, account)
+	if err != nil {
+		return nil, err
+	}
+
+	return &integrationstypes.GettableAWSAgentCheckIn{
+		AccountId:         account.ID.StringValue(),
+		CloudAccountId:    *account.AccountID,
+		RemovedAt:         account.RemovedAt,
+		IntegrationConfig: *agentConfig,
+	}, nil
+}
+
+func (a *awsProvider) getAWSAgentConfig(ctx context.Context, account *integrationstypes.CloudIntegration) (*integrationstypes.AWSAgentIntegrationConfig, error) {
+	// prepare and return integration config to be consumed by agent
+	agentConfig := &integrationstypes.AWSAgentIntegrationConfig{
+		EnabledRegions: []string{},
+		TelemetryCollectionStrategy: &integrationstypes.AWSCollectionStrategy{
+			Provider:   a.GetName(),
+			AWSMetrics: &integrationstypes.AWSMetricsStrategy{},
+			AWSLogs:    &integrationstypes.AWSLogsStrategy{},
+			S3Buckets:  map[string][]string{},
+		},
+	}
+
+	accountConfig := new(integrationstypes.AWSAccountConfig)
+	err := accountConfig.Unmarshal(account.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	if accountConfig != nil && accountConfig.EnabledRegions != nil {
+		agentConfig.EnabledRegions = accountConfig.EnabledRegions
+	}
+
+	svcConfigs, err := a.serviceConfigRepo.GetAllForAccount(
+		ctx, account.OrgID, account.ID.StringValue(),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// accumulate config in a fixed order to ensure same config generated across runs
+	configuredServices := maps.Keys(svcConfigs)
+	slices.Sort(configuredServices)
+
+	for _, svcType := range configuredServices {
+		definition, err := a.awsServiceDefinitions.GetServiceDefinition(ctx, svcType)
+		if err != nil {
+			continue
+		}
+		config := svcConfigs[svcType]
+
+		serviceConfig := new(integrationstypes.AWSCloudServiceConfig)
+		err = serviceConfig.Unmarshal(config)
+		if err != nil {
+			continue
+		}
+
+		if serviceConfig.Logs != nil && serviceConfig.Logs.Enabled {
+			if svcType == integrationstypes.S3Sync {
+				// S3 bucket sync; No cloudwatch logs are appended for this service type;
+				// Though definition is populated with a custom cloudwatch group that helps in calculating logs connection status
+				agentConfig.TelemetryCollectionStrategy.S3Buckets = serviceConfig.Logs.S3Buckets
+			} else if definition.Strategy.AWSLogs != nil { // services that includes a logs subscription
+				agentConfig.TelemetryCollectionStrategy.AWSLogs.Subscriptions = append(
+					agentConfig.TelemetryCollectionStrategy.AWSLogs.Subscriptions,
+					definition.Strategy.AWSLogs.Subscriptions...,
+				)
+			}
+		}
+
+		if serviceConfig.Metrics != nil && serviceConfig.Metrics.Enabled && definition.Strategy.AWSMetrics != nil {
+			agentConfig.TelemetryCollectionStrategy.AWSMetrics.StreamFilters = append(
+				agentConfig.TelemetryCollectionStrategy.AWSMetrics.StreamFilters,
+				definition.Strategy.AWSMetrics.StreamFilters...,
+			)
+		}
+	}
+
+	return agentConfig, nil
+}
+
+func (a *awsProvider) GetName() integrationstypes.CloudProviderType {
+	return integrationstypes.CloudProviderAWS
+}
+
+func (a *awsProvider) ListServices(ctx context.Context, orgID string, cloudAccountID *string) (any, error) {
+	svcConfigs := make(map[string]*integrationstypes.AWSCloudServiceConfig)
+	if cloudAccountID != nil {
+		activeAccount, err := a.accountsRepo.GetConnectedCloudAccount(ctx, orgID, a.GetName().String(), *cloudAccountID)
+		if err != nil {
+			return nil, err
+		}
+
+		serviceConfigs, err := a.serviceConfigRepo.GetAllForAccount(ctx, orgID, activeAccount.ID.String())
+		if err != nil {
+			return nil, err
+		}
+
+		for svcType, config := range serviceConfigs {
+			serviceConfig := new(integrationstypes.AWSCloudServiceConfig)
+			err = serviceConfig.Unmarshal(config)
+			if err != nil {
+				return nil, err
+			}
+			svcConfigs[svcType] = serviceConfig
+		}
+	}
+
+	summaries := make([]integrationstypes.AWSServiceSummary, 0)
+
+	definitions, err := a.awsServiceDefinitions.ListServiceDefinitions(ctx)
+	if err != nil {
+		return nil, model.InternalError(fmt.Errorf("couldn't list aws service definitions: %w", err))
+	}
+
+	for _, def := range definitions {
+		summary := integrationstypes.AWSServiceSummary{
+			DefinitionMetadata: def.DefinitionMetadata,
+			Config:             nil,
+		}
+
+		summary.Config = svcConfigs[summary.Id]
+
+		summaries = append(summaries, summary)
+	}
+
+	sort.Slice(summaries, func(i, j int) bool {
+		return summaries[i].DefinitionMetadata.Title < summaries[j].DefinitionMetadata.Title
+	})
+
+	return &integrationstypes.GettableAWSServices{
+		Services: summaries,
+	}, nil
+}
+
+func (a *awsProvider) GetServiceDetails(ctx context.Context, req *integrationstypes.GetServiceDetailsReq) (any, error) {
+	details := new(integrationstypes.GettableAWSServiceDetails)
+
+	awsDefinition, err := a.awsServiceDefinitions.GetServiceDefinition(ctx, req.ServiceId)
+	if err != nil {
+		return nil, model.InternalError(fmt.Errorf("couldn't get aws service definition: %w", err))
+	}
+
+	details.AWSServiceDefinition = *awsDefinition
+	details.Strategy.Provider = a.GetName()
+	if req.CloudAccountID == nil {
+		return details, nil
+	}
+
+	config, err := a.getServiceConfig(ctx, &details.AWSServiceDefinition, req.OrgID, a.GetName().String(), req.ServiceId, *req.CloudAccountID)
+	if err != nil {
+		return nil, err
+	}
+
+	if config == nil {
+		return details, nil
+	}
+
+	details.Config = config
+
+	connectionStatus, err := a.getServiceConnectionStatus(
+		ctx,
+		*req.CloudAccountID,
+		req.OrgID,
+		&details.AWSServiceDefinition,
+		config,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	details.ConnectionStatus = connectionStatus
+
+	return details, nil
+}
+
+func (a *awsProvider) getServiceConnectionStatus(
+	ctx context.Context,
+	cloudAccountID string,
+	orgID string,
+	def *integrationstypes.AWSServiceDefinition,
+	serviceConfig *integrationstypes.AWSCloudServiceConfig,
+) (*integrationstypes.ServiceConnectionStatus, error) {
+	if def.Strategy == nil {
+		return nil, nil
+	}
+
+	resp := new(integrationstypes.ServiceConnectionStatus)
+
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	if def.Strategy.AWSMetrics != nil && serviceConfig.Metrics.Enabled {
+		go func() {
+			defer func() {
+				if r := recover(); r != nil {
+					a.logger.ErrorContext(
+						ctx, "panic while getting service metrics connection status",
+						"error", r,
+						"service", def.DefinitionMetadata.Id,
+					)
+				}
+			}()
+			defer wg.Done()
+			status, _ := a.getServiceMetricsConnectionStatus(ctx, cloudAccountID, orgID, def)
+			resp.Metrics = status
+		}()
+	}
+
+	if def.Strategy.AWSLogs != nil && serviceConfig.Logs.Enabled {
+		go func() {
+			defer func() {
+				if r := recover(); r != nil {
+					a.logger.ErrorContext(
+						ctx, "panic while getting service logs connection status",
+						"error", r,
+						"service", def.DefinitionMetadata.Id,
+					)
+				}
+			}()
+			defer wg.Done()
+			status, _ := a.getServiceLogsConnectionStatus(ctx, cloudAccountID, orgID, def)
+			resp.Logs = status
+		}()
+	}
+
+	wg.Wait()
+
+	return resp, nil
+}
+
+func (a *awsProvider) getServiceMetricsConnectionStatus(
+	ctx context.Context,
+	cloudAccountID string,
+	orgID string,
+	def *integrationstypes.AWSServiceDefinition,
+) ([]*integrationstypes.SignalConnectionStatus, error) {
+	if def.Strategy == nil ||
+		len(def.Strategy.AWSMetrics.StreamFilters) < 1 ||
+		len(def.DataCollected.Metrics) < 1 {
+		return nil, nil
+	}
+
+	statusResp := make([]*integrationstypes.SignalConnectionStatus, 0)
+
+	for _, category := range def.IngestionStatusCheck.Metrics {
+		queries := make([]qbtypes.QueryEnvelope, 0)
+
+		for _, check := range category.Checks {
+			filterExpression := fmt.Sprintf(`cloud.provider="aws" AND cloud.account.id="%s"`, cloudAccountID)
+			f := ""
+			for _, attribute := range check.Attributes {
+				f = fmt.Sprintf("%s %s", attribute.Name, attribute.Operator)
+				if attribute.Value != "" {
+					f = fmt.Sprintf("%s '%s'", f, attribute.Value)
+				}
+
+				filterExpression = fmt.Sprintf("%s AND %s", filterExpression, f)
+			}
+
+			queries = append(queries, qbtypes.QueryEnvelope{
+				Type: qbtypes.QueryTypeBuilder,
+				Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
+					Name:   valuer.GenerateUUID().String(),
+					Signal: telemetrytypes.SignalMetrics,
+					Aggregations: []qbtypes.MetricAggregation{{
+						MetricName:       check.Key,
+						TimeAggregation:  metrictypes.TimeAggregationAvg,
+						SpaceAggregation: metrictypes.SpaceAggregationAvg,
+					}},
+					Filter: &qbtypes.Filter{
+						Expression: filterExpression,
+					},
+				},
+			})
+		}
+
+		resp, err := a.querier.QueryRange(ctx, valuer.MustNewUUID(orgID), &qbtypes.QueryRangeRequest{
+			SchemaVersion: "v5",
+			Start:         uint64(time.Now().Add(-time.Hour).UnixMilli()),
+			End:           uint64(time.Now().UnixMilli()),
+			RequestType:   qbtypes.RequestTypeScalar,
+			CompositeQuery: qbtypes.CompositeQuery{
+				Queries: queries,
+			},
+		})
+		if err != nil {
+			a.logger.DebugContext(ctx,
+				"error querying for service metrics connection status",
+				"error", err,
+				"service", def.DefinitionMetadata.Id,
+			)
+			continue
+		}
+
+		if resp != nil && len(resp.Data.Results) < 1 {
+			continue
+		}
+
+		queryResponse, ok := resp.Data.Results[0].(*qbtypes.TimeSeriesData)
+		if !ok {
+			continue
+		}
+
+		if queryResponse == nil ||
+			len(queryResponse.Aggregations) < 1 ||
+			len(queryResponse.Aggregations[0].Series) < 1 ||
+			len(queryResponse.Aggregations[0].Series[0].Values) < 1 {
+			continue
+		}
+
+		statusResp = append(statusResp, &integrationstypes.SignalConnectionStatus{
+			CategoryID:           category.Category,
+			CategoryDisplayName:  category.DisplayName,
+			LastReceivedTsMillis: queryResponse.Aggregations[0].Series[0].Values[0].Timestamp,
+			LastReceivedFrom:     "signoz-aws-integration",
+		})
+	}
+
+	return statusResp, nil
+}
+
+func (a *awsProvider) getServiceLogsConnectionStatus(
+	ctx context.Context,
+	cloudAccountID string,
+	orgID string,
+	def *integrationstypes.AWSServiceDefinition,
+) ([]*integrationstypes.SignalConnectionStatus, error) {
+	if def.Strategy == nil ||
+		len(def.Strategy.AWSLogs.Subscriptions) < 1 ||
+		len(def.DataCollected.Logs) < 1 {
+		return nil, nil
+	}
+
+	statusResp := make([]*integrationstypes.SignalConnectionStatus, 0)
+
+	for _, category := range def.IngestionStatusCheck.Logs {
+		queries := make([]qbtypes.QueryEnvelope, 0)
+
+		for _, check := range category.Checks {
+			filterExpression := fmt.Sprintf(`cloud.account.id="%s"`, cloudAccountID)
+			f := ""
+			for _, attribute := range check.Attributes {
+				f = fmt.Sprintf("%s %s", attribute.Name, attribute.Operator)
+				if attribute.Value != "" {
+					f = fmt.Sprintf("%s '%s'", f, attribute.Value)
+				}
+
+				filterExpression = fmt.Sprintf("%s AND %s", filterExpression, f)
+			}
+
+			queries = append(queries, qbtypes.QueryEnvelope{
+				Type: qbtypes.QueryTypeBuilder,
+				Spec: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+					Name:   valuer.GenerateUUID().String(),
+					Signal: telemetrytypes.SignalLogs,
+					Aggregations: []qbtypes.LogAggregation{{
+						Expression: "count()",
+					}},
+					Filter: &qbtypes.Filter{
+						Expression: filterExpression,
+					},
+					Limit:  10,
+					Offset: 0,
+				},
+			})
+		}
+
+		resp, err := a.querier.QueryRange(ctx, valuer.MustNewUUID(orgID), &qbtypes.QueryRangeRequest{
+			SchemaVersion: "v1",
+			Start:         uint64(time.Now().Add(-time.Hour * 1).UnixMilli()),
+			End:           uint64(time.Now().UnixMilli()),
+			RequestType:   qbtypes.RequestTypeTimeSeries,
+			CompositeQuery: qbtypes.CompositeQuery{
+				Queries: queries,
+			},
+		})
+		if err != nil {
+			a.logger.DebugContext(ctx,
+				"error querying for service logs connection status",
+				"error", err,
+				"service", def.DefinitionMetadata.Id,
+			)
+			continue
+		}
+
+		if resp != nil && len(resp.Data.Results) < 1 {
+			continue
+		}
+
+		queryResponse, ok := resp.Data.Results[0].(*qbtypes.TimeSeriesData)
+		if !ok {
+			continue
+		}
+
+		if queryResponse == nil ||
+			len(queryResponse.Aggregations) < 1 ||
+			len(queryResponse.Aggregations[0].Series) < 1 ||
+			len(queryResponse.Aggregations[0].Series[0].Values) < 1 {
+			continue
+		}
+
+		statusResp = append(statusResp, &integrationstypes.SignalConnectionStatus{
+			CategoryID:           category.Category,
+			CategoryDisplayName:  category.DisplayName,
+			LastReceivedTsMillis: queryResponse.Aggregations[0].Series[0].Values[0].Timestamp,
+			LastReceivedFrom:     "signoz-aws-integration",
+		})
+	}
+
+	return statusResp, nil
+}
+
+func (a *awsProvider) getServiceConfig(ctx context.Context,
+	def *integrationstypes.AWSServiceDefinition, orgID, cloudProvider, serviceId, cloudAccountId string,
+) (*integrationstypes.AWSCloudServiceConfig, error) {
+	activeAccount, err := a.accountsRepo.GetConnectedCloudAccount(ctx, orgID, cloudProvider, cloudAccountId)
+	if err != nil {
+		return nil, err
+	}
+
+	config, err := a.serviceConfigRepo.Get(ctx, orgID, activeAccount.ID.StringValue(), serviceId)
+	if err != nil {
+		if errors.Ast(err, errors.TypeNotFound) {
+			return nil, nil
+		}
+
+		return nil, err
+	}
+
+	serviceConfig := new(integrationstypes.AWSCloudServiceConfig)
+	err = serviceConfig.Unmarshal(config)
+	if err != nil {
+		return nil, err
+	}
+
+	if config != nil && serviceConfig.Metrics != nil && serviceConfig.Metrics.Enabled {
+		def.PopulateDashboardURLs(serviceId)
+	}
+
+	return serviceConfig, nil
+}
+
+func (a *awsProvider) GetAvailableDashboards(ctx context.Context, orgID valuer.UUID) ([]*dashboardtypes.Dashboard, error) {
+	accountRecords, err := a.accountsRepo.ListConnected(ctx, orgID.StringValue(), a.GetName().String())
+	if err != nil {
+		return nil, err
+	}
+
+	// for now service dashboards are only available when metrics are enabled.
+	servicesWithAvailableMetrics := map[string]*time.Time{}
+
+	for _, ar := range accountRecords {
+		if ar.AccountID != nil {
+			configsBySvcId, err := a.serviceConfigRepo.GetAllForAccount(ctx, orgID.StringValue(), ar.ID.StringValue())
+			if err != nil {
+				return nil, err
+			}
+
+			for svcId, config := range configsBySvcId {
+				serviceConfig := new(integrationstypes.AWSCloudServiceConfig)
+				err = serviceConfig.Unmarshal(config)
+				if err != nil {
+					return nil, err
+				}
+
+				if serviceConfig.Metrics != nil && serviceConfig.Metrics.Enabled {
+					servicesWithAvailableMetrics[svcId] = &ar.CreatedAt
+				}
+			}
+		}
+	}
+
+	svcDashboards := make([]*dashboardtypes.Dashboard, 0)
+
+	allServices, err := a.awsServiceDefinitions.ListServiceDefinitions(ctx)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to list aws service definitions")
+	}
+
+	for _, svc := range allServices {
+		serviceDashboardsCreatedAt, ok := servicesWithAvailableMetrics[svc.Id]
+		if ok {
+			svcDashboards = integrationstypes.GetDashboardsFromAssets(svc.Id, a.GetName(), serviceDashboardsCreatedAt, svc.Assets)
+			servicesWithAvailableMetrics[svc.Id] = nil
+		}
+	}
+
+	return svcDashboards, nil
+}
+
+func (a *awsProvider) GetDashboard(ctx context.Context, req *integrationstypes.GettableDashboard) (*dashboardtypes.Dashboard, error) {
+	allDashboards, err := a.GetAvailableDashboards(ctx, req.OrgID)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, d := range allDashboards {
+		if d.ID == req.ID {
+			return d, nil
+		}
+	}
+
+	return nil, errors.NewNotFoundf(CodeDashboardNotFound, "dashboard with id %s not found", req.ID)
+}
+
+func (a *awsProvider) GenerateConnectionArtifact(ctx context.Context, req *integrationstypes.PostableConnectionArtifact) (any, error) {
+	connection := new(integrationstypes.PostableAWSConnectionUrl)
+
+	err := connection.Unmarshal(req.Data)
+	if err != nil {
+		return nil, err
+	}
+
+	if connection.AccountConfig != nil {
+		for _, region := range connection.AccountConfig.EnabledRegions {
+			if integrationstypes.ValidAWSRegions[region] {
+				continue
+			}
+
+			return nil, errors.NewInvalidInputf(CodeInvalidAWSRegion, "invalid aws region: %s", region)
+		}
+	}
+
+	config, err := connection.AccountConfig.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	account, err := a.accountsRepo.Upsert(
+		ctx, req.OrgID, integrationstypes.CloudProviderAWS.String(), nil, config,
+		nil, nil, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	agentVersion := "v0.0.8"
+	if connection.AgentConfig.Version != "" {
+		agentVersion = connection.AgentConfig.Version
+	}
+
+	baseURL := fmt.Sprintf("https://%s.console.aws.amazon.com/cloudformation/home",
+		connection.AgentConfig.Region)
+	u, _ := url.Parse(baseURL)
+
+	q := u.Query()
+	q.Set("region", connection.AgentConfig.Region)
+	u.Fragment = "/stacks/quickcreate"
+
+	u.RawQuery = q.Encode()
+
+	q = u.Query()
+	q.Set("stackName", "signoz-integration")
+	q.Set("templateURL", fmt.Sprintf("https://signoz-integrations.s3.us-east-1.amazonaws.com/aws-quickcreate-template-%s.json", agentVersion))
+	q.Set("param_SigNozIntegrationAgentVersion", agentVersion)
+	q.Set("param_SigNozApiUrl", connection.AgentConfig.SigNozAPIUrl)
+	q.Set("param_SigNozApiKey", connection.AgentConfig.SigNozAPIKey)
+	q.Set("param_SigNozAccountId", account.ID.StringValue())
+	q.Set("param_IngestionUrl", connection.AgentConfig.IngestionUrl)
+	q.Set("param_IngestionKey", connection.AgentConfig.IngestionKey)
+
+	return &integrationstypes.GettableAWSConnectionUrl{
+		AccountId:     account.ID.StringValue(),
+		ConnectionUrl: u.String() + "?&" + q.Encode(), // this format is required by AWS
+	}, nil
+}
+
+func (a *awsProvider) UpdateServiceConfig(ctx context.Context, req *integrationstypes.PatchableServiceConfig) (any, error) {
+	definition, err := a.awsServiceDefinitions.GetServiceDefinition(ctx, req.ServiceId)
+	if err != nil {
+		return nil, model.InternalError(fmt.Errorf("couldn't get aws service definition: %w", err))
+	}
+
+	serviceConfig := new(integrationstypes.PatchableAWSCloudServiceConfig)
+	err = serviceConfig.Unmarshal(req.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = serviceConfig.Config.Validate(definition); err != nil {
+		return nil, err
+	}
+
+	// can only update config for a connected cloud account id
+	_, err = a.accountsRepo.GetConnectedCloudAccount(
+		ctx, req.OrgID, a.GetName().String(), serviceConfig.CloudAccountId,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceConfigBytes, err := serviceConfig.Config.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	updatedConfig, err := a.serviceConfigRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), serviceConfig.CloudAccountId, req.ServiceId, serviceConfigBytes,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = serviceConfig.Unmarshal(updatedConfig); err != nil {
+		return nil, err
+	}
+
+	return &integrationstypes.PatchServiceConfigResponse{
+		ServiceId: req.ServiceId,
+		Config:    serviceConfig,
+	}, nil
+}
+
+func (a *awsProvider) UpdateAccountConfig(ctx context.Context, req *integrationstypes.PatchableAccountConfig) (any, error) {
+	config := new(integrationstypes.PatchableAWSAccountConfig)
+
+	err := config.Unmarshal(req.Data)
+	if err != nil {
+		return nil, err
+	}
+
+	if config.Config == nil {
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "account config can't be null")
+	}
+
+	for _, region := range config.Config.EnabledRegions {
+		if integrationstypes.ValidAWSRegions[region] {
+			continue
+		}
+
+		return nil, errors.NewInvalidInputf(CodeInvalidAWSRegion, "invalid aws region: %s", region)
+	}
+
+	configBytes, err := config.Config.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	// account must exist to update config, but it doesn't need to be connected
+	_, err = a.accountsRepo.Get(ctx, req.OrgID, a.GetName().String(), req.AccountId)
+	if err != nil {
+		return nil, err
+	}
+
+	accountRecord, err := a.accountsRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), &req.AccountId, configBytes, nil, nil, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return accountRecord.Account(a.GetName()), nil
+}
+
+func (a *awsProvider) DisconnectAccount(ctx context.Context, orgID, accountID string) (*integrationstypes.CloudIntegration, error) {
+	account, err := a.accountsRepo.Get(ctx, orgID, a.GetName().String(), accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	tsNow := time.Now()
+	account, err = a.accountsRepo.Upsert(
+		ctx, orgID, a.GetName().String(), &accountID, nil, nil, nil, &tsNow,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return account, nil
+}

pkg/query-service/app/cloudintegrations/implazureprovider/azureprovider.go

@@ -0,0 +1,561 @@
+package implazureprovider
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"slices"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/querier"
+	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/services"
+	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/store"
+	"github.com/SigNoz/signoz/pkg/query-service/model"
+	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"golang.org/x/exp/maps"
+)
+
+var (
+	CodeInvalidAzureRegion = errors.MustNewCode("invalid_azure_region")
+	CodeDashboardNotFound  = errors.MustNewCode("dashboard_not_found")
+)
+
+type azureProvider struct {
+	logger                  *slog.Logger
+	accountsRepo            store.CloudProviderAccountsRepository
+	serviceConfigRepo       store.ServiceConfigDatabase
+	azureServiceDefinitions *services.AzureServicesProvider
+	querier                 querier.Querier
+}
+
+func NewAzureCloudProvider(
+	logger *slog.Logger,
+	accountsRepo store.CloudProviderAccountsRepository,
+	serviceConfigRepo store.ServiceConfigDatabase,
+	querier querier.Querier,
+) integrationstypes.CloudProvider {
+	azureServiceDefinitions, err := services.NewAzureCloudProviderServices()
+	if err != nil {
+		panic("failed to initialize Azure service definitions: " + err.Error())
+	}
+
+	return &azureProvider{
+		logger:                  logger,
+		accountsRepo:            accountsRepo,
+		serviceConfigRepo:       serviceConfigRepo,
+		azureServiceDefinitions: azureServiceDefinitions,
+		querier:                 querier,
+	}
+}
+
+func (a *azureProvider) GetAccountStatus(ctx context.Context, orgID, accountID string) (*integrationstypes.GettableAccountStatus, error) {
+	account, err := a.accountsRepo.Get(ctx, orgID, a.GetName().String(), accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	return &integrationstypes.GettableAccountStatus{
+		Id:             account.ID.String(),
+		CloudAccountId: account.AccountID,
+		Status:         account.Status(),
+	}, nil
+}
+
+func (a *azureProvider) ListConnectedAccounts(ctx context.Context, orgID string) (*integrationstypes.GettableConnectedAccountsList, error) {
+	accountRecords, err := a.accountsRepo.ListConnected(ctx, orgID, a.GetName().String())
+	if err != nil {
+		return nil, err
+	}
+
+	connectedAccounts := make([]*integrationstypes.Account, 0, len(accountRecords))
+	for _, r := range accountRecords {
+		connectedAccounts = append(connectedAccounts, r.Account(a.GetName()))
+	}
+
+	return &integrationstypes.GettableConnectedAccountsList{
+		Accounts: connectedAccounts,
+	}, nil
+}
+
+func (a *azureProvider) AgentCheckIn(ctx context.Context, req *integrationstypes.PostableAgentCheckInPayload) (any, error) {
+	existingAccount, err := a.accountsRepo.Get(ctx, req.OrgID, a.GetName().String(), req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	if existingAccount != nil && existingAccount.AccountID != nil && *existingAccount.AccountID != req.AccountID {
+		return nil, model.BadRequest(fmt.Errorf(
+			"can't check in with new %s account id %s for account %s with existing %s id %s",
+			a.GetName().String(), req.AccountID, existingAccount.ID.StringValue(), a.GetName().String(),
+			*existingAccount.AccountID,
+		))
+	}
+
+	existingAccount, err = a.accountsRepo.GetConnectedCloudAccount(ctx, req.OrgID, a.GetName().String(), req.AccountID)
+	if existingAccount != nil && existingAccount.ID.StringValue() != req.ID {
+		return nil, model.BadRequest(fmt.Errorf(
+			"can't check in to %s account %s with id %s. already connected with id %s",
+			a.GetName().String(), req.AccountID, req.ID, existingAccount.ID.StringValue(),
+		))
+	}
+
+	agentReport := integrationstypes.AgentReport{
+		TimestampMillis: time.Now().UnixMilli(),
+		Data:            req.Data,
+	}
+
+	account, err := a.accountsRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), &req.ID, nil, &req.AccountID, &agentReport, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	agentConfig, err := a.getAzureAgentConfig(ctx, account)
+	if err != nil {
+		return nil, err
+	}
+
+	return &integrationstypes.GettableAzureAgentCheckIn{
+		AccountId:         account.ID.StringValue(),
+		CloudAccountId:    *account.AccountID,
+		RemovedAt:         account.RemovedAt,
+		IntegrationConfig: *agentConfig,
+	}, nil
+}
+
+func (a *azureProvider) getAzureAgentConfig(ctx context.Context, account *integrationstypes.CloudIntegration) (*integrationstypes.AzureAgentIntegrationConfig, error) {
+	// prepare and return integration config to be consumed by agent
+	agentConfig := &integrationstypes.AzureAgentIntegrationConfig{
+		TelemetryCollectionStrategy: &integrationstypes.AzureCollectionStrategy{
+			Provider:     a.GetName(),
+			AzureMetrics: make([]*integrationstypes.AzureMetricsStrategy, 0),
+			AzureLogs:    make([]*integrationstypes.AzureLogsStrategy, 0),
+		},
+	}
+
+	accountConfig := new(integrationstypes.AzureAccountConfig)
+	err := accountConfig.Unmarshal(account.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	if account.Config != nil {
+		agentConfig.DeploymentRegion = accountConfig.DeploymentRegion
+		agentConfig.EnabledResourceGroups = accountConfig.EnabledResourceGroups
+	}
+
+	svcConfigs, err := a.serviceConfigRepo.GetAllForAccount(
+		ctx, account.OrgID, account.ID.StringValue(),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// accumulate config in a fixed order to ensure same config generated across runs
+	configuredServices := maps.Keys(svcConfigs)
+	slices.Sort(configuredServices)
+
+	metrics := make([]*integrationstypes.AzureMetricsStrategy, 0)
+	logs := make([]*integrationstypes.AzureLogsStrategy, 0)
+
+	for _, svcType := range configuredServices {
+		definition, err := a.azureServiceDefinitions.GetServiceDefinition(ctx, svcType)
+		if err != nil {
+			continue
+		}
+		config := svcConfigs[svcType]
+
+		serviceConfig := new(integrationstypes.AzureCloudServiceConfig)
+		err = serviceConfig.Unmarshal(config)
+		if err != nil {
+			continue
+		}
+
+		metricsStrategyMap := make(map[string]*integrationstypes.AzureMetricsStrategy)
+		logsStrategyMap := make(map[string]*integrationstypes.AzureLogsStrategy)
+
+		for _, metric := range definition.Strategy.AzureMetrics {
+			metricsStrategyMap[metric.Name] = metric
+		}
+
+		for _, log := range definition.Strategy.AzureLogs {
+			logsStrategyMap[log.Name] = log
+		}
+
+		if serviceConfig.Metrics != nil {
+			for _, metric := range serviceConfig.Metrics {
+				if metric.Enabled {
+					metrics = append(metrics, &integrationstypes.AzureMetricsStrategy{
+						CategoryType: metricsStrategyMap[metric.Name].CategoryType,
+						Name:         metric.Name,
+					})
+				}
+			}
+		}
+
+		if serviceConfig.Logs != nil {
+			for _, log := range serviceConfig.Logs {
+				if log.Enabled {
+					logs = append(logs, &integrationstypes.AzureLogsStrategy{
+						CategoryType: logsStrategyMap[log.Name].CategoryType,
+						Name:         log.Name,
+					})
+				}
+			}
+		}
+	}
+
+	agentConfig.TelemetryCollectionStrategy.AzureMetrics = metrics
+	agentConfig.TelemetryCollectionStrategy.AzureLogs = logs
+
+	return agentConfig, nil
+}
+
+func (a *azureProvider) GetName() valuer.String {
+	return integrationstypes.CloudProviderAzure
+}
+
+func (a *azureProvider) ListServices(ctx context.Context, orgID string, cloudAccountID *string) (any, error) {
+	svcConfigs := make(map[string]*integrationstypes.AzureCloudServiceConfig)
+	if cloudAccountID != nil {
+		activeAccount, err := a.accountsRepo.GetConnectedCloudAccount(ctx, orgID, a.GetName().String(), *cloudAccountID)
+		if err != nil {
+			return nil, err
+		}
+
+		serviceConfigs, err := a.serviceConfigRepo.GetAllForAccount(ctx, orgID, activeAccount.ID.StringValue())
+		if err != nil {
+			return nil, err
+		}
+
+		for svcType, config := range serviceConfigs {
+			serviceConfig := new(integrationstypes.AzureCloudServiceConfig)
+			err = serviceConfig.Unmarshal(config)
+			if err != nil {
+				return nil, err
+			}
+			svcConfigs[svcType] = serviceConfig
+		}
+	}
+
+	summaries := make([]integrationstypes.AzureServiceSummary, 0)
+
+	definitions, err := a.azureServiceDefinitions.ListServiceDefinitions(ctx)
+	if err != nil {
+		return nil, model.InternalError(fmt.Errorf("couldn't list aws service definitions: %w", err))
+	}
+
+	for _, def := range definitions {
+		summary := integrationstypes.AzureServiceSummary{
+			DefinitionMetadata: def.DefinitionMetadata,
+			Config:             nil,
+		}
+
+		summary.Config = svcConfigs[summary.Id]
+
+		summaries = append(summaries, summary)
+	}
+
+	sort.Slice(summaries, func(i, j int) bool {
+		return summaries[i].DefinitionMetadata.Title < summaries[j].DefinitionMetadata.Title
+	})
+
+	return &integrationstypes.GettableAzureServices{
+		Services: summaries,
+	}, nil
+}
+
+func (a *azureProvider) GetServiceDetails(ctx context.Context, req *integrationstypes.GetServiceDetailsReq) (any, error) {
+	details := new(integrationstypes.GettableAzureServiceDetails)
+
+	azureDefinition, err := a.azureServiceDefinitions.GetServiceDefinition(ctx, req.ServiceId)
+	if err != nil {
+		return nil, model.InternalError(fmt.Errorf("couldn't get aws service definition: %w", err))
+	}
+
+	details.AzureServiceDefinition = *azureDefinition
+	details.Strategy.Provider = a.GetName()
+	if req.CloudAccountID == nil {
+		return details, nil
+	}
+
+	config, err := a.getServiceConfig(ctx, &details.AzureServiceDefinition, req.OrgID, req.ServiceId, *req.CloudAccountID)
+	if err != nil {
+		return nil, err
+	}
+
+	if config == nil {
+		return details, nil
+	}
+
+	details.Config = config
+
+	return details, nil
+}
+
+func (a *azureProvider) getServiceConfig(
+	ctx context.Context,
+	definition *integrationstypes.AzureServiceDefinition,
+	orgID string,
+	serviceId string,
+	cloudAccountId string,
+) (*integrationstypes.AzureCloudServiceConfig, error) {
+	activeAccount, err := a.accountsRepo.GetConnectedCloudAccount(ctx, orgID, a.GetName().String(), cloudAccountId)
+	if err != nil {
+		return nil, err
+	}
+
+	configBytes, err := a.serviceConfigRepo.Get(ctx, orgID, activeAccount.ID.String(), serviceId)
+	if err != nil {
+		if errors.Ast(err, errors.TypeNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	config := new(integrationstypes.AzureCloudServiceConfig)
+	err = config.Unmarshal(configBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, metric := range config.Metrics {
+		if metric.Enabled {
+			definition.PopulateDashboardURLs(serviceId)
+			break
+		}
+	}
+
+	return config, nil
+}
+
+func (a *azureProvider) GenerateConnectionArtifact(ctx context.Context, req *integrationstypes.PostableConnectionArtifact) (any, error) {
+	connection := new(integrationstypes.PostableAzureConnectionCommand)
+
+	err := connection.Unmarshal(req.Data)
+	if err != nil {
+		return nil, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "failed unmarshal request data into AWS connection config")
+	}
+
+	// validate connection config
+	if connection.AccountConfig != nil {
+		if !integrationstypes.ValidAzureRegions[connection.AccountConfig.DeploymentRegion] {
+			return nil, errors.NewInvalidInputf(CodeInvalidAzureRegion, "invalid azure region: %s",
+				connection.AccountConfig.DeploymentRegion,
+			)
+		}
+	}
+
+	config, err := connection.AccountConfig.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	account, err := a.accountsRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), nil, config,
+		nil, nil, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	//agentVersion := "v0.0.1"
+	//
+	//if req.AgentConfig.Version != "" {
+	//	agentVersion = req.AgentConfig.Version
+	//}
+
+	// TODO: improve cli command generation
+	cliCommand := []string{"az", "stack", "sub", "create", "--name", "SigNozIntegration", "--location",
+		connection.AccountConfig.DeploymentRegion, "--template-uri", "https://raw.githubusercontent.com/swagftw/signoz-pocs/refs/heads/main/template.json",
+		"--action-on-unmanage", "deleteAll", "--deny-settings-mode", "denyDelete", "--parameters", fmt.Sprintf("rgName=%s", "signoz-integration-rg"),
+		fmt.Sprintf("rgLocation=%s", connection.AccountConfig.DeploymentRegion)}
+
+	return &integrationstypes.GettableAzureConnectionCommand{
+		AccountId:                   account.ID.String(),
+		AzureShellConnectionCommand: "az create",
+		AzureCliConnectionCommand:   strings.Join(cliCommand, " "),
+	}, nil
+}
+
+func (a *azureProvider) UpdateServiceConfig(ctx context.Context, req *integrationstypes.PatchableServiceConfig) (any, error) {
+	definition, err := a.azureServiceDefinitions.GetServiceDefinition(ctx, req.ServiceId)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceConfig := new(integrationstypes.PatchableAzureCloudServiceConfig)
+	err = serviceConfig.Unmarshal(req.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = serviceConfig.Config.Validate(definition); err != nil {
+		return nil, err
+	}
+
+	// can only update config for a connected cloud account id
+	_, err = a.accountsRepo.GetConnectedCloudAccount(
+		ctx, req.OrgID, a.GetName().String(), serviceConfig.CloudAccountId,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceConfigBytes, err := serviceConfig.Config.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	updatedConfig, err := a.serviceConfigRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), serviceConfig.CloudAccountId, req.ServiceId, serviceConfigBytes,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = serviceConfig.Unmarshal(updatedConfig); err != nil {
+		return nil, err
+	}
+
+	return &integrationstypes.PatchServiceConfigResponse{
+		ServiceId: req.ServiceId,
+		Config:    serviceConfig,
+	}, nil
+}
+
+func (a *azureProvider) GetAvailableDashboards(ctx context.Context, orgID valuer.UUID) ([]*dashboardtypes.Dashboard, error) {
+	accountRecords, err := a.accountsRepo.ListConnected(ctx, orgID.StringValue(), a.GetName().String())
+	if err != nil {
+		return nil, err
+	}
+
+	// for now service dashboards are only available when metrics are enabled.
+	servicesWithAvailableMetrics := map[string]*time.Time{}
+
+	for _, ar := range accountRecords {
+		if ar.AccountID == nil {
+			continue
+		}
+
+		configsBySvcId, err := a.serviceConfigRepo.GetAllForAccount(ctx, orgID.StringValue(), ar.ID.StringValue())
+		if err != nil {
+			return nil, err
+		}
+
+		for svcId, config := range configsBySvcId {
+			serviceConfig := new(integrationstypes.AzureCloudServiceConfig)
+			err = serviceConfig.Unmarshal(config)
+			if err != nil {
+				return nil, err
+			}
+
+			if serviceConfig.Metrics != nil {
+				for _, metric := range serviceConfig.Metrics {
+					if metric.Enabled {
+						servicesWithAvailableMetrics[svcId] = &ar.CreatedAt
+						break
+					}
+				}
+			}
+		}
+	}
+
+	svcDashboards := make([]*dashboardtypes.Dashboard, 0)
+
+	allServices, err := a.azureServiceDefinitions.ListServiceDefinitions(ctx)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to list azure service definitions")
+	}
+
+	for _, svc := range allServices {
+		serviceDashboardsCreatedAt := servicesWithAvailableMetrics[svc.Id]
+		if serviceDashboardsCreatedAt != nil {
+			svcDashboards = integrationstypes.GetDashboardsFromAssets(svc.Id, a.GetName(), serviceDashboardsCreatedAt, svc.Assets)
+			servicesWithAvailableMetrics[svc.Id] = nil
+		}
+	}
+
+	return svcDashboards, nil
+}
+
+func (a *azureProvider) GetDashboard(ctx context.Context, req *integrationstypes.GettableDashboard) (*dashboardtypes.Dashboard, error) {
+	allDashboards, err := a.GetAvailableDashboards(ctx, req.OrgID)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, dashboard := range allDashboards {
+		if dashboard.ID == req.ID {
+			return dashboard, nil
+		}
+	}
+
+	return nil, errors.NewNotFoundf(CodeDashboardNotFound, "dashboard with id %s not found", req.ID)
+}
+
+func (a *azureProvider) UpdateAccountConfig(ctx context.Context, req *integrationstypes.PatchableAccountConfig) (any, error) {
+	config := new(integrationstypes.PatchableAzureAccountConfig)
+
+	err := config.Unmarshal(req.Data)
+	if err != nil {
+		return nil, err
+	}
+
+	if config.Config == nil && len(config.Config.EnabledResourceGroups) < 1 {
+		return nil, errors.NewInvalidInputf(CodeInvalidAzureRegion, "azure region and resource groups must be provided")
+	}
+
+	//for azure, preserve deployment region if already set
+	account, err := a.accountsRepo.Get(ctx, req.OrgID, a.GetName().String(), req.AccountId)
+	if err != nil {
+		return nil, err
+	}
+
+	storedConfig := new(integrationstypes.AzureAccountConfig)
+	err = storedConfig.Unmarshal(account.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	if account.Config != nil {
+		config.Config.DeploymentRegion = storedConfig.DeploymentRegion
+	}
+
+	configBytes, err := config.Config.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	accountRecord, err := a.accountsRepo.Upsert(
+		ctx, req.OrgID, a.GetName().String(), &req.AccountId, configBytes, nil, nil, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return accountRecord.Account(a.GetName()), nil
+}
+
+func (a *azureProvider) DisconnectAccount(ctx context.Context, orgID, accountID string) (*integrationstypes.CloudIntegration, error) {
+	account, err := a.accountsRepo.Get(ctx, orgID, a.GetName().String(), accountID)
+	if err != nil {
+		return nil, err
+	}
+
+	tsNow := time.Now()
+	account, err = a.accountsRepo.Upsert(
+		ctx, orgID, a.GetName().String(), &accountID, nil, nil, nil, &tsNow,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return account, nil
+}

pkg/query-service/app/cloudintegrations/models.go

@@ -1,94 +1 @@
 package cloudintegrations
-
-import (
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/services"
-	"github.com/SigNoz/signoz/pkg/types"
-)
-
-type ServiceSummary struct {
-	services.Metadata
-
-	Config *types.CloudServiceConfig `json:"config"`
-}
-
-type ServiceDetails struct {
-	services.Definition
-
-	Config           *types.CloudServiceConfig `json:"config"`
-	ConnectionStatus *ServiceConnectionStatus  `json:"status,omitempty"`
-}
-
-type AccountStatus struct {
-	Integration AccountIntegrationStatus `json:"integration"`
-}
-
-type AccountIntegrationStatus struct {
-	LastHeartbeatTsMillis *int64 `json:"last_heartbeat_ts_ms"`
-}
-
-type LogsConfig struct {
-	Enabled   bool                `json:"enabled"`
-	S3Buckets map[string][]string `json:"s3_buckets,omitempty"`
-}
-
-type MetricsConfig struct {
-	Enabled bool `json:"enabled"`
-}
-
-type ServiceConnectionStatus struct {
-	Logs    *SignalConnectionStatus `json:"logs"`
-	Metrics *SignalConnectionStatus `json:"metrics"`
-}
-
-type SignalConnectionStatus struct {
-	LastReceivedTsMillis int64  `json:"last_received_ts_ms"` // epoch milliseconds
-	LastReceivedFrom     string `json:"last_received_from"`  // resource identifier
-}
-
-type CompiledCollectionStrategy = services.CollectionStrategy
-
-func NewCompiledCollectionStrategy(provider string) (*CompiledCollectionStrategy, error) {
-	if provider == "aws" {
-		return &CompiledCollectionStrategy{
-			Provider:   "aws",
-			AWSMetrics: &services.AWSMetricsStrategy{},
-			AWSLogs:    &services.AWSLogsStrategy{},
-		}, nil
-	}
-	return nil, errors.NewNotFoundf(services.CodeUnsupportedCloudProvider, "unsupported cloud provider: %s", provider)
-}
-
-// Helper for accumulating strategies for enabled services.
-func AddServiceStrategy(serviceType string, cs *CompiledCollectionStrategy,
-	definitionStrat *services.CollectionStrategy, config *types.CloudServiceConfig) error {
-	if definitionStrat.Provider != cs.Provider {
-		return errors.NewInternalf(CodeMismatchCloudProvider, "can't add %s service strategy to compiled strategy for %s",
-			definitionStrat.Provider, cs.Provider)
-	}
-
-	if cs.Provider == "aws" {
-		if config.Logs != nil && config.Logs.Enabled {
-			if serviceType == services.S3Sync {
-				// S3 bucket sync; No cloudwatch logs are appended for this service type;
-				// Though definition is populated with a custom cloudwatch group that helps in calculating logs connection status
-				cs.S3Buckets = config.Logs.S3Buckets
-			} else if definitionStrat.AWSLogs != nil { // services that includes a logs subscription
-				cs.AWSLogs.Subscriptions = append(
-					cs.AWSLogs.Subscriptions,
-					definitionStrat.AWSLogs.Subscriptions...,
-				)
-			}
-		}
-		if config.Metrics != nil && config.Metrics.Enabled && definitionStrat.AWSMetrics != nil {
-			cs.AWSMetrics.StreamFilters = append(
-				cs.AWSMetrics.StreamFilters,
-				definitionStrat.AWSMetrics.StreamFilters...,
-			)
-		}
-
-		return nil
-	}
-
-	return errors.NewNotFoundf(services.CodeUnsupportedCloudProvider, "unsupported cloud provider: %s", cs.Provider)
-}

pkg/query-service/app/cloudintegrations/providerregistry.go

@@ -0,0 +1,26 @@
+package cloudintegrations
+
+import (
+	"log/slog"
+
+	"github.com/SigNoz/signoz/pkg/querier"
+	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/implawsprovider"
+	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/implazureprovider"
+	integrationstore "github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/store"
+	"github.com/SigNoz/signoz/pkg/sqlstore"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
+)
+
+func NewCloudProviderRegistry(logger *slog.Logger, store sqlstore.SQLStore, querier querier.Querier) map[integrationstypes.CloudProviderType]integrationstypes.CloudProvider {
+	registry := make(map[integrationstypes.CloudProviderType]integrationstypes.CloudProvider)
+
+	accountsRepo := integrationstore.NewCloudProviderAccountsRepository(store)
+	serviceConfigRepo := integrationstore.NewServiceConfigRepository(store)
+
+	awsProviderImpl := implawsprovider.NewAWSCloudProvider(logger, accountsRepo, serviceConfigRepo, querier)
+	registry[integrationstypes.CloudProviderAWS] = awsProviderImpl
+	azureProviderImpl := implazureprovider.NewAzureCloudProvider(logger, accountsRepo, serviceConfigRepo, querier)
+	registry[integrationstypes.CloudProviderAzure] = azureProviderImpl
+
+	return registry
+}

pkg/query-service/app/cloudintegrations/services/definitions/aws/alb/integration.json

@@ -7,6 +7,24 @@
     "metrics": true,
     "logs": false
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_ApplicationELB_ConsumedLCUs_count",
+            "attributes": []
+          },
+          {
+            "key": "aws_ApplicationELB_ProcessedBytes_sum",
+            "attributes": []
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {

pkg/query-service/app/cloudintegrations/services/definitions/aws/api-gateway/integration.json

@@ -7,6 +7,75 @@
     "metrics": true,
     "logs": true
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "rest_api",
+        "display_name": "REST API Metrics",
+        "checks": [
+          {
+            "key": "aws_ApiGateway_Count_count",
+            "attributes": [
+              {
+                "name": "ApiName",
+                "operator": "EXISTS",
+                "value": ""
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "category": "http_api",
+        "display_name": "HTTP API Metrics",
+        "checks": [
+          {
+            "key": "aws_ApiGateway_Count_count",
+            "attributes": [
+              {
+                "name": "ApiId",
+                "operator": "EXISTS",
+                "value": ""
+              }
+            ]
+          }
+        ]
+      },
+      {
+        "category": "websocket_api",
+        "display_name": "Websocket API Metrics",
+        "checks": [
+          {
+            "key": "aws_ApiGateway_Count_count",
+            "attributes": [
+              {
+                "name": "ApiId",
+                "operator": "EXISTS",
+                "value": ""
+              }
+            ]
+          }
+        ]
+      }
+    ],
+    "logs": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "attributes": [
+              {
+                "name": "aws.cloudwatch.log_group_name",
+                "operator": "ILIKE",
+                "value": "API-Gateway%"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {
@@ -148,6 +217,146 @@
         "name": "aws_ApiGateway_Latency_sum",
         "unit": "Milliseconds",
         "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_4xx_sum",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_4xx_max",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_4xx_min",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_4xx_count",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_5xx_sum",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_5xx_max",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_5xx_min",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_5xx_count",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_DataProcessed_sum",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_DataProcessed_max",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_DataProcessed_min",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_DataProcessed_count",
+        "unit": "Bytes",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ExecutionError_sum",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ExecutionError_max",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ExecutionError_min",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ExecutionError_count",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ClientError_sum",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ClientError_max",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ClientError_min",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ClientError_count",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_IntegrationError_sum",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_IntegrationError_max",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_IntegrationError_min",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_IntegrationError_count",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ConnectCount_sum",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ConnectCount_max",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ConnectCount_min",
+        "unit": "Count",
+        "type": "Gauge"
+      },
+      {
+        "name": "aws_ApiGateway_ConnectCount_count",
+        "unit": "Count",
+        "type": "Gauge"
       }
     ],
     "logs": [

pkg/query-service/app/cloudintegrations/services/definitions/aws/dynamodb/integration.json

@@ -7,6 +7,24 @@
     "metrics": true,
     "logs": false
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_DynamoDB_AccountMaxReads_max",
+            "attributes": []
+          },
+          {
+            "key": "aws_DynamoDB_AccountProvisionedReadCapacityUtilization_max",
+            "attributes": []
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {
@@ -391,4 +409,4 @@
       }
     ]
   }
-}
+}

pkg/query-service/app/cloudintegrations/services/definitions/aws/ec2/integration.json

@@ -7,6 +7,24 @@
     "metrics": true,
     "logs": false
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_EC2_CPUUtilization_max",
+            "attributes": []
+          },
+          {
+            "key": "aws_EC2_NetworkIn_max",
+            "attributes": []
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {
@@ -515,4 +533,4 @@
       }
     ]
   }
-}
+}

pkg/query-service/app/cloudintegrations/services/definitions/aws/ecs/integration.json

@@ -7,6 +7,81 @@
     "metrics": true,
     "logs": true
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "overview",
+        "display_name": "Overview",
+        "checks": [
+          {
+            "key": "aws_ECS_CPUUtilization_max",
+            "attributes": []
+          },
+          {
+            "key": "aws_ECS_MemoryUtilization_max",
+            "attributes": []
+          }
+        ]
+      },
+      {
+        "category": "containerinsights",
+        "display_name": "Container Insights",
+        "checks": [
+          {
+            "key": "aws_ECS_ContainerInsights_NetworkRxBytes_max",
+            "attributes": []
+          },
+          {
+            "key": "aws_ECS_ContainerInsights_StorageReadBytes_max",
+            "attributes": []
+          }
+        ]
+      },
+      {
+        "category": "enhanced_containerinsights",
+        "display_name": "Enhanced Container Insights",
+        "checks": [
+          {
+            "key": "aws_ECS_ContainerInsights_ContainerCpuUtilization_max",
+            "attributes": [
+              {
+                "name": "TaskId",
+                "operator": "EXISTS",
+                "value": ""
+              }
+            ]
+          },
+          {
+            "key": "aws_ECS_ContainerInsights_TaskMemoryUtilization_max",
+            "attributes": [
+              {
+                "name": "TaskId",
+                "operator": "EXISTS",
+                "value": ""
+              }
+            ]
+          }
+        ]
+      }
+    ],
+    "logs": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "attributes": [
+              {
+                "name": "aws.cloudwatch.log_group_name",
+                "operator": "ILIKE",
+                "value": "%/ecs/%"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {

pkg/query-service/app/cloudintegrations/services/definitions/aws/elasticache/integration.json

@@ -7,6 +7,20 @@
     "metrics": true,
     "logs": false
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_ElastiCache_CacheHitRate_max",
+            "attributes": []
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics":[
       {
@@ -1928,7 +1942,7 @@
         "unit": "Percent",
         "type": "Gauge",
         "description": ""
-      }                      
+      }
     ]
   },
   "telemetry_collection_strategy": {
@@ -1951,4 +1965,4 @@
       }
     ]
   }
-}
+}

pkg/query-service/app/cloudintegrations/services/definitions/aws/lambda/integration.json

@@ -7,6 +7,37 @@
     "metrics": true,
     "logs": true
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_Lambda_Invocations_sum",
+            "attributes": []
+          }
+        ]
+      }
+    ],
+    "logs": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "attributes": [
+              {
+                "name": "aws.cloudwatch.log_group_name",
+                "operator": "ILIKE",
+                "value": "/aws/lambda%"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {

pkg/query-service/app/cloudintegrations/services/definitions/aws/msk/integration.json

@@ -7,6 +7,20 @@
       "metrics": true,
       "logs": false
     },
+    "ingestion_status_check": {
+      "metrics": [
+        {
+          "category": "$default",
+          "display_name": "Default",
+          "checks": [
+            {
+              "key": "aws_Kafka_KafkaDataLogsDiskUsed_max",
+              "attributes": []
+            }
+          ]
+        }
+      ]
+    },
     "data_collected": {
       "metrics": [
         {
@@ -1088,4 +1102,3 @@
       ]
     }
   }
-  

pkg/query-service/app/cloudintegrations/services/definitions/aws/rds/integration.json

@@ -7,6 +7,37 @@
     "metrics": true,
     "logs": true
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_RDS_CPUUtilization_max",
+            "attributes": []
+          }
+        ]
+      }
+    ],
+    "logs": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "attributes": [
+              {
+                "name": "resources.aws.cloudwatch.log_group_name",
+                "operator": "ILIKE",
+                "value": "/aws/rds%"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {
@@ -800,4 +831,4 @@
       }
     ]
   }
-}
+}

pkg/query-service/app/cloudintegrations/services/definitions/aws/sns/integration.json

@@ -7,6 +7,20 @@
     "metrics": true,
     "logs": false
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_SNS_NumberOfMessagesPublished_sum",
+            "attributes": []
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {
@@ -127,4 +141,4 @@
       }
     ]
   }
-}
+}

pkg/query-service/app/cloudintegrations/services/definitions/aws/sqs/integration.json

@@ -7,6 +7,24 @@
     "metrics": true,
     "logs": false
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "key": "aws_SQS_SentMessageSize_max",
+            "attributes": []
+          },
+          {
+            "key": "aws_SQS_NumberOfMessagesSent_sum",
+            "attributes": []
+          }
+        ]
+      }
+    ]
+  },
   "data_collected": {
     "metrics": [
       {
@@ -247,4 +265,4 @@
       }
     ]
   }
-}
+}

pkg/query-service/app/cloudintegrations/services/definitions/azure/blobstorage/assets/dashboards/overview.json

@@ -0,0 +1,3 @@
+{
+
+}

pkg/query-service/app/cloudintegrations/services/definitions/azure/blobstorage/icon.svg

@@ -0,0 +1 @@
+<svg id="f2f04349-8aee-4413-84c9-a9053611b319" xmlns="http://www.w3.org/2000/svg" width="18" height="18" viewBox="0 0 18 18"><defs><linearGradient id="ad4c4f96-09aa-4f91-ba10-5cb8ad530f74" x1="9" y1="15.83" x2="9" y2="5.79" gradientUnits="userSpaceOnUse"><stop offset="0" stop-color="#b3b3b3" /><stop offset="0.26" stop-color="#c1c1c1" /><stop offset="1" stop-color="#e6e6e6" /></linearGradient></defs><title>Icon-storage-86</title><path d="M.5,5.79h17a0,0,0,0,1,0,0v9.48a.57.57,0,0,1-.57.57H1.07a.57.57,0,0,1-.57-.57V5.79A0,0,0,0,1,.5,5.79Z" fill="url(#ad4c4f96-09aa-4f91-ba10-5cb8ad530f74)" /><path d="M1.07,2.17H16.93a.57.57,0,0,1,.57.57V5.79a0,0,0,0,1,0,0H.5a0,0,0,0,1,0,0V2.73A.57.57,0,0,1,1.07,2.17Z" fill="#37c2b1" /><path d="M2.81,6.89H15.18a.27.27,0,0,1,.26.27v1.4a.27.27,0,0,1-.26.27H2.81a.27.27,0,0,1-.26-.27V7.16A.27.27,0,0,1,2.81,6.89Z" fill="#fff" /><path d="M2.82,9.68H15.19a.27.27,0,0,1,.26.27v1.41a.27.27,0,0,1-.26.27H2.82a.27.27,0,0,1-.26-.27V10A.27.27,0,0,1,2.82,9.68Z" fill="#37c2b1" /><path d="M2.82,12.5H15.19a.27.27,0,0,1,.26.27v1.41a.27.27,0,0,1-.26.27H2.82a.27.27,0,0,1-.26-.27V12.77A.27.27,0,0,1,2.82,12.5Z" fill="#258277" /></svg>

pkg/query-service/app/cloudintegrations/services/definitions/azure/blobstorage/integration.json

@@ -0,0 +1,63 @@
+{
+  "id": "blobstorage",
+  "title": "Blob Storage",
+  "icon": "file://icon.svg",
+  "overview": "file://overview.md",
+  "supported_signals": {
+    "metrics": true,
+    "logs": true
+  },
+  "data_collected": {
+    "metrics": [
+      {
+        "name": "placeholder_metric_1",
+        "unit": "Percent",
+        "type": "Gauge",
+        "description": ""
+      }
+    ],
+    "logs": [
+      {
+        "name": "placeholder_log_1",
+        "path": "placeholder.path.value",
+        "type": "string"
+      }
+    ]
+  },
+  "telemetry_collection_strategy": {
+    "azure_metrics": [
+      {
+        "category_type": "metrics",
+        "name": "Capacity"
+      },
+      {
+        "category_type": "metrics",
+        "name": "Transaction"
+      }
+    ],
+    "azure_logs": [
+      {
+        "category_type": "logs",
+        "name": "StorageRead"
+      },
+      {
+        "category_type": "logs",
+        "name": "StorageWrite"
+      },
+      {
+        "category_type": "logs",
+        "name": "StorageDelete"
+      }
+    ]
+  },
+  "assets": {
+    "dashboards": [
+      {
+        "id": "overview",
+        "title": "Blob Storage Overview",
+        "description": "Overview of Blob Storage",
+        "definition": "file://assets/dashboards/overview.json"
+      }
+    ]
+  }
+}

pkg/query-service/app/cloudintegrations/services/models.go

@@ -1,91 +1 @@
 package services
-
-import (
-	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
-)
-
-type Metadata struct {
-	Id    string `json:"id"`
-	Title string `json:"title"`
-	Icon  string `json:"icon"`
-}
-
-type Definition struct {
-	Metadata
-
-	Overview string `json:"overview"` // markdown
-
-	Assets Assets `json:"assets"`
-
-	SupportedSignals SupportedSignals `json:"supported_signals"`
-
-	DataCollected DataCollected `json:"data_collected"`
-
-	Strategy *CollectionStrategy `json:"telemetry_collection_strategy"`
-}
-
-type Assets struct {
-	Dashboards []Dashboard `json:"dashboards"`
-}
-
-type SupportedSignals struct {
-	Logs    bool `json:"logs"`
-	Metrics bool `json:"metrics"`
-}
-
-type DataCollected struct {
-	Logs    []CollectedLogAttribute `json:"logs"`
-	Metrics []CollectedMetric       `json:"metrics"`
-}
-
-type CollectedLogAttribute struct {
-	Name string `json:"name"`
-	Path string `json:"path"`
-	Type string `json:"type"`
-}
-
-type CollectedMetric struct {
-	Name        string `json:"name"`
-	Type        string `json:"type"`
-	Unit        string `json:"unit"`
-	Description string `json:"description"`
-}
-
-type CollectionStrategy struct {
-	Provider string `json:"provider"`
-
-	AWSMetrics *AWSMetricsStrategy `json:"aws_metrics,omitempty"`
-	AWSLogs    *AWSLogsStrategy    `json:"aws_logs,omitempty"`
-	S3Buckets  map[string][]string `json:"s3_buckets,omitempty"` // Only available in S3 Sync Service Type
-}
-
-type AWSMetricsStrategy struct {
-	// to be used as https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-cloudwatch-metricstream.html#cfn-cloudwatch-metricstream-includefilters
-	StreamFilters []struct {
-		// json tags here are in the shape expected by AWS API as detailed at
-		// https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-cloudwatch-metricstream-metricstreamfilter.html
-		Namespace   string   `json:"Namespace"`
-		MetricNames []string `json:"MetricNames,omitempty"`
-	} `json:"cloudwatch_metric_stream_filters"`
-}
-
-type AWSLogsStrategy struct {
-	Subscriptions []struct {
-		// subscribe to all logs groups with specified prefix.
-		// eg: `/aws/rds/`
-		LogGroupNamePrefix string `json:"log_group_name_prefix"`
-
-		// https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/FilterAndPatternSyntax.html
-		// "" implies no filtering is required.
-		FilterPattern string `json:"filter_pattern"`
-	} `json:"cloudwatch_logs_subscriptions"`
-}
-
-type Dashboard struct {
-	Id          string                                `json:"id"`
-	Url         string                                `json:"url"`
-	Title       string                                `json:"title"`
-	Description string                                `json:"description"`
-	Image       string                                `json:"image"`
-	Definition  *dashboardtypes.StorableDashboardData `json:"definition,omitempty"`
-}

pkg/query-service/app/cloudintegrations/services/services.go

@@ -2,128 +2,128 @@ package services
 
 import (
 	"bytes"
+	"context"
 	"embed"
 	"encoding/json"
 	"fmt"
 	"io/fs"
 	"path"
-	"sort"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/query-service/app/integrations"
-	"github.com/SigNoz/signoz/pkg/query-service/model"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
 	koanfJson "github.com/knadh/koanf/parsers/json"
-	"golang.org/x/exp/maps"
-)
-
-const (
-	S3Sync = "s3sync"
 )
 
 var (
-	CodeUnsupportedCloudProvider = errors.MustNewCode("unsupported_cloud_provider")
-	CodeUnsupportedServiceType   = errors.MustNewCode("unsupported_service_type")
+	CodeServiceDefinitionNotFound = errors.MustNewCode("service_definition_not_dound")
 )
 
-func List(cloudProvider string) ([]Definition, *model.ApiError) {
-	cloudServices, found := supportedServices[cloudProvider]
-	if !found || cloudServices == nil {
-		return nil, model.NotFoundError(fmt.Errorf(
-			"unsupported cloud provider: %s", cloudProvider,
-		))
+type (
+	AWSServicesProvider struct {
+		definitions map[string]*integrationstypes.AWSServiceDefinition
 	}
+	AzureServicesProvider struct {
+		definitions map[string]*integrationstypes.AzureServiceDefinition
+	}
+)
 
-	services := maps.Values(cloudServices)
-	sort.Slice(services, func(i, j int) bool {
-		return services[i].Id < services[j].Id
-	})
-
-	return services, nil
+func (a *AzureServicesProvider) ListServiceDefinitions(ctx context.Context) (map[string]*integrationstypes.AzureServiceDefinition, error) {
+	return a.definitions, nil
 }
 
-func Map(cloudProvider string) (map[string]Definition, error) {
-	cloudServices, found := supportedServices[cloudProvider]
-	if !found || cloudServices == nil {
-		return nil, errors.Newf(errors.TypeNotFound, CodeUnsupportedCloudProvider, "unsupported cloud provider: %s", cloudProvider)
+func (a *AzureServicesProvider) GetServiceDefinition(ctx context.Context, serviceName string) (*integrationstypes.AzureServiceDefinition, error) {
+	def, ok := a.definitions[serviceName]
+	if !ok {
+		return nil, errors.NewNotFoundf(CodeServiceDefinitionNotFound, "azure service definition not found: %s", serviceName)
+	}
+
+	return def, nil
+}
+
+func (a *AWSServicesProvider) ListServiceDefinitions(ctx context.Context) (map[string]*integrationstypes.AWSServiceDefinition, error) {
+	return a.definitions, nil
+}
+
+func (a *AWSServicesProvider) GetServiceDefinition(ctx context.Context, serviceName string) (*integrationstypes.AWSServiceDefinition, error) {
+	def, ok := a.definitions[serviceName]
+	if !ok {
+		return nil, errors.NewNotFoundf(CodeServiceDefinitionNotFound, "aws service definition not found: %s", serviceName)
+	}
+
+	return def, nil
+}
+
+func NewAWSCloudProviderServices() (*AWSServicesProvider, error) {
+	definitions, err := readAllServiceDefinitions(integrationstypes.CloudProviderAWS)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceDefinitions := make(map[string]*integrationstypes.AWSServiceDefinition)
+	for id, def := range definitions {
+		typedDef, ok := def.(*integrationstypes.AWSServiceDefinition)
+		if !ok {
+			return nil, fmt.Errorf("invalid type for AWS service definition %s", id)
+		}
+		serviceDefinitions[id] = typedDef
+	}
+
+	return &AWSServicesProvider{
+		definitions: serviceDefinitions,
+	}, nil
+}
+
+func NewAzureCloudProviderServices() (*AzureServicesProvider, error) {
+	definitions, err := readAllServiceDefinitions(integrationstypes.CloudProviderAzure)
+	if err != nil {
+		return nil, err
+	}
+
+	serviceDefinitions := make(map[string]*integrationstypes.AzureServiceDefinition)
+	for id, def := range definitions {
+		typedDef, ok := def.(*integrationstypes.AzureServiceDefinition)
+		if !ok {
+			return nil, fmt.Errorf("invalid type for Azure service definition %s", id)
+		}
+		serviceDefinitions[id] = typedDef
+	}
+
+	return &AzureServicesProvider{
+		definitions: serviceDefinitions,
+	}, nil
+}
+
+// End of API. Logic for reading service definition files follows
+
+//go:embed definitions/*
+var definitionFiles embed.FS
+
+func readAllServiceDefinitions(cloudProvider valuer.String) (map[string]any, error) {
+	rootDirName := "definitions"
+
+	cloudProviderDirPath := path.Join(rootDirName, cloudProvider.String())
+
+	cloudServices, err := readServiceDefinitionsFromDir(cloudProvider, cloudProviderDirPath)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(cloudServices) < 1 {
+		return nil, errors.NewInternalf(errors.CodeInternal, "no service definitions found in %s", cloudProviderDirPath)
 	}
 
 	return cloudServices, nil
 }
 
-func GetServiceDefinition(cloudProvider, serviceType string) (*Definition, error) {
-	cloudServices := supportedServices[cloudProvider]
-	if cloudServices == nil {
-		return nil, errors.Newf(errors.TypeNotFound, CodeUnsupportedCloudProvider, "unsupported cloud provider: %s", cloudProvider)
-	}
-
-	svc, exists := cloudServices[serviceType]
-	if !exists {
-		return nil, errors.Newf(errors.TypeNotFound, CodeUnsupportedServiceType, "%s service not found: %s", cloudProvider, serviceType)
-	}
-
-	return &svc, nil
-}
-
-// End of API. Logic for reading service definition files follows
-
-// Service details read from ./serviceDefinitions
-// { "providerName": { "service_id": {...}} }
-var supportedServices map[string]map[string]Definition
-
-func init() {
-	err := readAllServiceDefinitions()
-	if err != nil {
-		panic(fmt.Errorf(
-			"couldn't read cloud service definitions: %w", err,
-		))
-	}
-}
-
-//go:embed definitions/*
-var definitionFiles embed.FS
-
-func readAllServiceDefinitions() error {
-	supportedServices = map[string]map[string]Definition{}
-
-	rootDirName := "definitions"
-
-	cloudProviderDirs, err := fs.ReadDir(definitionFiles, rootDirName)
-	if err != nil {
-		return fmt.Errorf("couldn't read dirs in %s: %w", rootDirName, err)
-	}
-
-	for _, d := range cloudProviderDirs {
-		if !d.IsDir() {
-			continue
-		}
-
-		cloudProvider := d.Name()
-
-		cloudProviderDirPath := path.Join(rootDirName, cloudProvider)
-		cloudServices, err := readServiceDefinitionsFromDir(cloudProvider, cloudProviderDirPath)
-		if err != nil {
-			return fmt.Errorf("couldn't read %s service definitions: %w", cloudProvider, err)
-		}
-
-		if len(cloudServices) < 1 {
-			return fmt.Errorf("no %s services could be read", cloudProvider)
-		}
-
-		supportedServices[cloudProvider] = cloudServices
-	}
-
-	return nil
-}
-
-func readServiceDefinitionsFromDir(cloudProvider string, cloudProviderDirPath string) (
-	map[string]Definition, error,
-) {
+func readServiceDefinitionsFromDir(cloudProvider valuer.String, cloudProviderDirPath string) (map[string]any, error) {
 	svcDefDirs, err := fs.ReadDir(definitionFiles, cloudProviderDirPath)
 	if err != nil {
-		return nil, fmt.Errorf("couldn't list integrations dirs: %w", err)
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't list integrations dirs")
 	}
 
-	svcDefs := map[string]Definition{}
+	svcDefs := make(map[string]any)
 
 	for _, d := range svcDefDirs {
 		if !d.IsDir() {
@@ -133,103 +133,73 @@ func readServiceDefinitionsFromDir(cloudProvider string, cloudProviderDirPath st
 		svcDirPath := path.Join(cloudProviderDirPath, d.Name())
 		s, err := readServiceDefinition(cloudProvider, svcDirPath)
 		if err != nil {
-			return nil, fmt.Errorf("couldn't read svc definition for %s: %w", d.Name(), err)
+			return nil, err
 		}
 
-		_, exists := svcDefs[s.Id]
+		_, exists := svcDefs[s.GetId()]
 		if exists {
-			return nil, fmt.Errorf(
-				"duplicate service definition for id %s at %s", s.Id, d.Name(),
-			)
+			return nil, errors.NewInternalf(errors.CodeInternal, "duplicate service definition for id %s at %s", s.GetId(), d.Name())
 		}
-		svcDefs[s.Id] = *s
+		svcDefs[s.GetId()] = s
 	}
 
 	return svcDefs, nil
 }
 
-func readServiceDefinition(cloudProvider string, svcDirpath string) (*Definition, error) {
+func readServiceDefinition(cloudProvider valuer.String, svcDirpath string) (integrationstypes.Definition, error) {
 	integrationJsonPath := path.Join(svcDirpath, "integration.json")
 
 	serializedSpec, err := definitionFiles.ReadFile(integrationJsonPath)
 	if err != nil {
-		return nil, fmt.Errorf(
-			"couldn't find integration.json in %s: %w",
-			svcDirpath, err,
-		)
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't read integration definition in %s", svcDirpath)
 	}
 
 	integrationSpec, err := koanfJson.Parser().Unmarshal(serializedSpec)
 	if err != nil {
-		return nil, fmt.Errorf(
-			"couldn't parse integration.json from %s: %w",
-			integrationJsonPath, err,
-		)
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't parse integration definition in %s", svcDirpath)
 	}
 
-	hydrated, err := integrations.HydrateFileUris(
-		integrationSpec, definitionFiles, svcDirpath,
-	)
+	hydrated, err := integrations.HydrateFileUris(integrationSpec, definitionFiles, svcDirpath)
 	if err != nil {
-		return nil, fmt.Errorf(
-			"couldn't hydrate files referenced in service definition %s: %w",
-			integrationJsonPath, err,
-		)
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't hydrate integration definition in %s", svcDirpath)
 	}
 	hydratedSpec := hydrated.(map[string]any)
 
-	serviceDef, err := ParseStructWithJsonTagsFromMap[Definition](hydratedSpec)
-	if err != nil {
-		return nil, fmt.Errorf(
-			"couldn't parse hydrated JSON spec read from %s: %w",
-			integrationJsonPath, err,
-		)
+	var serviceDef integrationstypes.Definition
+
+	switch cloudProvider {
+	case integrationstypes.CloudProviderAWS:
+		serviceDef = &integrationstypes.AWSServiceDefinition{}
+	case integrationstypes.CloudProviderAzure:
+		serviceDef = &integrationstypes.AzureServiceDefinition{}
+	default:
+		// ideally this shouldn't happen hence throwing internal error
+		return nil, errors.NewInternalf(errors.CodeInternal, "unsupported cloud provider: %s", cloudProvider)
 	}
 
-	err = validateServiceDefinition(serviceDef)
+	err = parseStructWithJsonTagsFromMap(hydratedSpec, serviceDef)
 	if err != nil {
-		return nil, fmt.Errorf("invalid service definition %s: %w", serviceDef.Id, err)
+		return nil, err
+	}
+	err = serviceDef.Validate()
+	if err != nil {
+		return nil, err
 	}
-
-	serviceDef.Strategy.Provider = cloudProvider
 
 	return serviceDef, nil
-
 }
 
-func validateServiceDefinition(s *Definition) error {
-	// Validate dashboard data
-	seenDashboardIds := map[string]interface{}{}
-	for _, dd := range s.Assets.Dashboards {
-		if _, seen := seenDashboardIds[dd.Id]; seen {
-			return fmt.Errorf("multiple dashboards found with id %s", dd.Id)
-		}
-		seenDashboardIds[dd.Id] = nil
-	}
-
-	if s.Strategy == nil {
-		return fmt.Errorf("telemetry_collection_strategy is required")
-	}
-
-	// potentially more to follow
-
-	return nil
-}
-
-func ParseStructWithJsonTagsFromMap[StructType any](data map[string]any) (
-	*StructType, error,
-) {
+func parseStructWithJsonTagsFromMap(data map[string]any, target interface{}) error {
 	mapJson, err := json.Marshal(data)
 	if err != nil {
-		return nil, fmt.Errorf("couldn't marshal map to json: %w", err)
+		return errors.WrapInternalf(err, errors.CodeInternal, "couldn't marshal service definition json data")
 	}
 
-	var res StructType
 	decoder := json.NewDecoder(bytes.NewReader(mapJson))
 	decoder.DisallowUnknownFields()
-	err = decoder.Decode(&res)
+	err = decoder.Decode(target)
 	if err != nil {
-		return nil, fmt.Errorf("couldn't unmarshal json back to struct: %w", err)
+		return errors.WrapInternalf(err, errors.CodeInternal, "couldn't unmarshal service definition json data")
 	}
-	return &res, nil
+	return nil
 }

pkg/query-service/app/cloudintegrations/services/services_test.go

@@ -1,35 +1,3 @@
 package services
 
-import (
-	"testing"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/query-service/model"
-	"github.com/stretchr/testify/require"
-)
-
-func TestAvailableServices(t *testing.T) {
-	require := require.New(t)
-
-	// should be able to list available services.
-	_, apiErr := List("bad-cloud-provider")
-	require.NotNil(apiErr)
-	require.Equal(model.ErrorNotFound, apiErr.Type())
-
-	awsSvcs, apiErr := List("aws")
-	require.Nil(apiErr)
-	require.Greater(len(awsSvcs), 0)
-
-	// should be able to get details of a service
-	_, err := GetServiceDefinition(
-		"aws", "bad-service-id",
-	)
-	require.NotNil(err)
-	require.True(errors.Ast(err, errors.TypeNotFound))
-
-	svc, err := GetServiceDefinition(
-		"aws", awsSvcs[0].Id,
-	)
-	require.Nil(err)
-	require.Equal(*svc, awsSvcs[0])
-}
+// TODO: add more tests for services package

pkg/query-service/app/cloudintegrations/store/accountsRepo.go

@@ -1,4 +1,4 @@
-package cloudintegrations
+package store
 
 import (
 	"context"
@@ -7,49 +7,50 @@ import (
 	"strings"
 	"time"
 
-	"github.com/SigNoz/signoz/pkg/query-service/model"
+	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
-type cloudProviderAccountsRepository interface {
-	listConnected(ctx context.Context, orgId string, provider string) ([]types.CloudIntegration, *model.ApiError)
+var (
+	CodeCloudIntegrationAccountNotFound errors.Code = errors.MustNewCode("cloud_integration_account_not_found")
+)
 
-	get(ctx context.Context, orgId string, provider string, id string) (*types.CloudIntegration, *model.ApiError)
+type CloudProviderAccountsRepository interface {
+	ListConnected(ctx context.Context, orgId string, provider string) ([]integrationstypes.CloudIntegration, error)
 
-	getConnectedCloudAccount(ctx context.Context, orgId string, provider string, accountID string) (*types.CloudIntegration, *model.ApiError)
+	Get(ctx context.Context, orgId string, provider string, id string) (*integrationstypes.CloudIntegration, error)
+
+	GetConnectedCloudAccount(ctx context.Context, orgId string, provider string, accountID string) (*integrationstypes.CloudIntegration, error)
 
 	// Insert an account or update it by (cloudProvider, id)
 	// for specified non-empty fields
-	upsert(
+	Upsert(
 		ctx context.Context,
 		orgId string,
 		provider string,
 		id *string,
-		config *types.AccountConfig,
+		config []byte,
 		accountId *string,
-		agentReport *types.AgentReport,
+		agentReport *integrationstypes.AgentReport,
 		removedAt *time.Time,
-	) (*types.CloudIntegration, *model.ApiError)
+	) (*integrationstypes.CloudIntegration, error)
 }
 
-func newCloudProviderAccountsRepository(store sqlstore.SQLStore) (
-	*cloudProviderAccountsSQLRepository, error,
-) {
-	return &cloudProviderAccountsSQLRepository{
-		store: store,
-	}, nil
+func NewCloudProviderAccountsRepository(store sqlstore.SQLStore) CloudProviderAccountsRepository {
+	return &cloudProviderAccountsSQLRepository{store: store}
 }
 
 type cloudProviderAccountsSQLRepository struct {
 	store sqlstore.SQLStore
 }
 
-func (r *cloudProviderAccountsSQLRepository) listConnected(
+func (r *cloudProviderAccountsSQLRepository) ListConnected(
 	ctx context.Context, orgId string, cloudProvider string,
-) ([]types.CloudIntegration, *model.ApiError) {
-	accounts := []types.CloudIntegration{}
+) ([]integrationstypes.CloudIntegration, error) {
+	accounts := []integrationstypes.CloudIntegration{}
 
 	err := r.store.BunDB().NewSelect().
 		Model(&accounts).
@@ -62,18 +63,16 @@ func (r *cloudProviderAccountsSQLRepository) listConnected(
 		Scan(ctx)
 
 	if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"could not query connected cloud accounts: %w", err,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "could not query connected cloud accounts")
 	}
 
 	return accounts, nil
 }
 
-func (r *cloudProviderAccountsSQLRepository) get(
+func (r *cloudProviderAccountsSQLRepository) Get(
 	ctx context.Context, orgId string, provider string, id string,
-) (*types.CloudIntegration, *model.ApiError) {
-	var result types.CloudIntegration
+) (*integrationstypes.CloudIntegration, error) {
+	var result integrationstypes.CloudIntegration
 
 	err := r.store.BunDB().NewSelect().
 		Model(&result).
@@ -82,23 +81,25 @@ func (r *cloudProviderAccountsSQLRepository) get(
 		Where("id = ?", id).
 		Scan(ctx)
 
-	if err == sql.ErrNoRows {
-		return nil, model.NotFoundError(fmt.Errorf(
-			"couldn't find account with Id %s", id,
-		))
-	} else if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"couldn't query cloud provider accounts: %w", err,
-		))
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, errors.WrapNotFoundf(
+				err,
+				CodeCloudIntegrationAccountNotFound,
+				"couldn't find account with Id %s", id,
+			)
+		}
+
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't query cloud provider account")
 	}
 
 	return &result, nil
 }
 
-func (r *cloudProviderAccountsSQLRepository) getConnectedCloudAccount(
+func (r *cloudProviderAccountsSQLRepository) GetConnectedCloudAccount(
 	ctx context.Context, orgId string, provider string, accountId string,
-) (*types.CloudIntegration, *model.ApiError) {
-	var result types.CloudIntegration
+) (*integrationstypes.CloudIntegration, error) {
+	var result integrationstypes.CloudIntegration
 
 	err := r.store.BunDB().NewSelect().
 		Model(&result).
@@ -109,29 +110,25 @@ func (r *cloudProviderAccountsSQLRepository) getConnectedCloudAccount(
 		Where("removed_at is NULL").
 		Scan(ctx)
 
-	if err == sql.ErrNoRows {
-		return nil, model.NotFoundError(fmt.Errorf(
-			"couldn't find connected cloud account %s", accountId,
-		))
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, errors.WrapNotFoundf(err, CodeCloudIntegrationAccountNotFound, "couldn't find connected cloud account %s", accountId)
 	} else if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"couldn't query cloud provider accounts: %w", err,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't query cloud provider account")
 	}
 
 	return &result, nil
 }
 
-func (r *cloudProviderAccountsSQLRepository) upsert(
+func (r *cloudProviderAccountsSQLRepository) Upsert(
 	ctx context.Context,
 	orgId string,
 	provider string,
 	id *string,
-	config *types.AccountConfig,
+	config []byte,
 	accountId *string,
-	agentReport *types.AgentReport,
+	agentReport *integrationstypes.AgentReport,
 	removedAt *time.Time,
-) (*types.CloudIntegration, *model.ApiError) {
+) (*integrationstypes.CloudIntegration, error) {
 	// Insert
 	if id == nil {
 		temp := valuer.GenerateUUID().StringValue()
@@ -181,7 +178,7 @@ func (r *cloudProviderAccountsSQLRepository) upsert(
 		)
 	}
 
-	integration := types.CloudIntegration{
+	integration := integrationstypes.CloudIntegration{
 		OrgID:        orgId,
 		Provider:     provider,
 		Identifiable: types.Identifiable{ID: valuer.MustNewUUID(*id)},
@@ -195,22 +192,18 @@ func (r *cloudProviderAccountsSQLRepository) upsert(
 		RemovedAt:       removedAt,
 	}
 
-	_, dbErr := r.store.BunDB().NewInsert().
+	_, err := r.store.BunDB().NewInsert().
 		Model(&integration).
 		On(onConflictClause).
 		Exec(ctx)
 
-	if dbErr != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"could not upsert cloud account record: %w", dbErr,
-		))
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't upsert cloud integration account")
 	}
 
-	upsertedAccount, apiErr := r.get(ctx, orgId, provider, *id)
-	if apiErr != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"couldn't fetch upserted account by id: %w", apiErr.ToError(),
-		))
+	upsertedAccount, err := r.Get(ctx, orgId, provider, *id)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't get upserted cloud integration account")
 	}
 
 	return upsertedAccount, nil

pkg/query-service/app/cloudintegrations/store/serviceconfig_db.go

@@ -1,64 +1,63 @@
-package cloudintegrations
+package store
 
 import (
 	"context"
 	"database/sql"
-	"fmt"
 	"time"
 
-	"github.com/SigNoz/signoz/pkg/query-service/model"
+	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
+var (
+	CodeServiceConfigNotFound = errors.MustNewCode("service_config_not_found")
+)
+
 type ServiceConfigDatabase interface {
-	get(
+	Get(
 		ctx context.Context,
 		orgID string,
 		cloudAccountId string,
 		serviceType string,
-	) (*types.CloudServiceConfig, *model.ApiError)
+	) ([]byte, error)
 
-	upsert(
+	Upsert(
 		ctx context.Context,
 		orgID string,
 		cloudProvider string,
 		cloudAccountId string,
 		serviceId string,
-		config types.CloudServiceConfig,
-	) (*types.CloudServiceConfig, *model.ApiError)
+		config []byte,
+	) ([]byte, error)
 
-	getAllForAccount(
+	GetAllForAccount(
 		ctx context.Context,
 		orgID string,
 		cloudAccountId string,
 	) (
-		configsBySvcId map[string]*types.CloudServiceConfig,
-		apiErr *model.ApiError,
+		map[string][]byte,
+		error,
 	)
 }
 
-func newServiceConfigRepository(store sqlstore.SQLStore) (
-	*serviceConfigSQLRepository, error,
-) {
-	return &serviceConfigSQLRepository{
-		store: store,
-	}, nil
+func NewServiceConfigRepository(store sqlstore.SQLStore) ServiceConfigDatabase {
+	return &serviceConfigSQLRepository{store: store}
 }
 
 type serviceConfigSQLRepository struct {
 	store sqlstore.SQLStore
 }
 
-func (r *serviceConfigSQLRepository) get(
+func (r *serviceConfigSQLRepository) Get(
 	ctx context.Context,
 	orgID string,
 	cloudAccountId string,
 	serviceType string,
-) (*types.CloudServiceConfig, *model.ApiError) {
-
-	var result types.CloudIntegrationService
+) ([]byte, error) {
+	var result integrationstypes.CloudIntegrationService
 
 	err := r.store.BunDB().NewSelect().
 		Model(&result).
@@ -67,36 +66,30 @@ func (r *serviceConfigSQLRepository) get(
 		Where("ci.id = ?", cloudAccountId).
 		Where("cis.type = ?", serviceType).
 		Scan(ctx)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, errors.WrapNotFoundf(err, CodeServiceConfigNotFound, "couldn't find config for cloud account %s", cloudAccountId)
+		}
 
-	if err == sql.ErrNoRows {
-		return nil, model.NotFoundError(fmt.Errorf(
-			"couldn't find config for cloud account %s",
-			cloudAccountId,
-		))
-	} else if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"couldn't query cloud service config: %w", err,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't query cloud service config")
 	}
 
-	return &result.Config, nil
-
+	return result.Config, nil
 }
 
-func (r *serviceConfigSQLRepository) upsert(
+func (r *serviceConfigSQLRepository) Upsert(
 	ctx context.Context,
 	orgID string,
 	cloudProvider string,
 	cloudAccountId string,
 	serviceId string,
-	config types.CloudServiceConfig,
-) (*types.CloudServiceConfig, *model.ApiError) {
-
+	config []byte,
+) ([]byte, error) {
 	// get cloud integration id from account id
 	// if the account is not connected, we don't need to upsert the config
 	var cloudIntegrationId string
 	err := r.store.BunDB().NewSelect().
-		Model((*types.CloudIntegration)(nil)).
+		Model((*integrationstypes.CloudIntegration)(nil)).
 		Column("id").
 		Where("provider = ?", cloudProvider).
 		Where("account_id = ?", cloudAccountId).
@@ -104,14 +97,18 @@ func (r *serviceConfigSQLRepository) upsert(
 		Where("removed_at is NULL").
 		Where("last_agent_report is not NULL").
 		Scan(ctx, &cloudIntegrationId)
-
 	if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"couldn't query cloud integration id: %w", err,
-		))
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, errors.WrapNotFoundf(
+				err,
+				CodeCloudIntegrationAccountNotFound,
+				"couldn't find active cloud integration account",
+			)
+		}
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't query cloud integration id")
 	}
 
-	serviceConfig := types.CloudIntegrationService{
+	serviceConfig := integrationstypes.CloudIntegrationService{
 		Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
 		TimeAuditable: types.TimeAuditable{
 			CreatedAt: time.Now(),
@@ -126,21 +123,18 @@ func (r *serviceConfigSQLRepository) upsert(
 		On("conflict(cloud_integration_id, type) do update set config=excluded.config, updated_at=excluded.updated_at").
 		Exec(ctx)
 	if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"could not upsert cloud service config: %w", err,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't upsert cloud service config")
 	}
 
-	return &serviceConfig.Config, nil
-
+	return config, nil
 }
 
-func (r *serviceConfigSQLRepository) getAllForAccount(
+func (r *serviceConfigSQLRepository) GetAllForAccount(
 	ctx context.Context,
 	orgID string,
 	cloudAccountId string,
-) (map[string]*types.CloudServiceConfig, *model.ApiError) {
-	serviceConfigs := []types.CloudIntegrationService{}
+) (map[string][]byte, error) {
+	var serviceConfigs []integrationstypes.CloudIntegrationService
 
 	err := r.store.BunDB().NewSelect().
 		Model(&serviceConfigs).
@@ -149,15 +143,13 @@ func (r *serviceConfigSQLRepository) getAllForAccount(
 		Where("ci.org_id = ?", orgID).
 		Scan(ctx)
 	if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"could not query service configs from db: %w", err,
-		))
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't query service configs from db")
 	}
 
-	result := map[string]*types.CloudServiceConfig{}
+	result := make(map[string][]byte)
 
 	for _, r := range serviceConfigs {
-		result[r.Type] = &r.Config
+		result[r.Type] = r.Config
 	}
 
 	return result, nil

pkg/query-service/app/integrations/manager.go

@@ -11,6 +11,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types"
 	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
 	ruletypes "github.com/SigNoz/signoz/pkg/types/ruletypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
@@ -107,7 +108,7 @@ type IntegrationsListItem struct {
 
 type Integration struct {
 	IntegrationDetails
-	Installation *types.InstalledIntegration `json:"installation"`
+	Installation *integrationstypes.InstalledIntegration `json:"installation"`
 }
 
 type Manager struct {
@@ -223,7 +224,7 @@ func (m *Manager) InstallIntegration(
 	ctx context.Context,
 	orgId string,
 	integrationId string,
-	config types.InstalledIntegrationConfig,
+	config integrationstypes.InstalledIntegrationConfig,
 ) (*IntegrationsListItem, *model.ApiError) {
 	integrationDetails, apiErr := m.getIntegrationDetails(ctx, integrationId)
 	if apiErr != nil {
@@ -429,7 +430,7 @@ func (m *Manager) getInstalledIntegration(
 	ctx context.Context,
 	orgId string,
 	integrationId string,
-) (*types.InstalledIntegration, *model.ApiError) {
+) (*integrationstypes.InstalledIntegration, *model.ApiError) {
 	iis, apiErr := m.installedIntegrationsRepo.get(
 		ctx, orgId, []string{integrationId},
 	)
@@ -457,7 +458,7 @@ func (m *Manager) getInstalledIntegrations(
 		return nil, apiErr
 	}
 
-	installedTypes := utils.MapSlice(installations, func(i types.InstalledIntegration) string {
+	installedTypes := utils.MapSlice(installations, func(i integrationstypes.InstalledIntegration) string {
 		return i.Type
 	})
 	integrationDetails, apiErr := m.availableIntegrationsRepo.get(ctx, installedTypes)

pkg/query-service/app/integrations/repo.go

@@ -4,22 +4,22 @@ import (
 	"context"
 
 	"github.com/SigNoz/signoz/pkg/query-service/model"
-	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 )
 
 type InstalledIntegrationsRepo interface {
-	list(ctx context.Context, orgId string) ([]types.InstalledIntegration, *model.ApiError)
+	list(ctx context.Context, orgId string) ([]integrationstypes.InstalledIntegration, *model.ApiError)
 
 	get(
 		ctx context.Context, orgId string, integrationTypes []string,
-	) (map[string]types.InstalledIntegration, *model.ApiError)
+	) (map[string]integrationstypes.InstalledIntegration, *model.ApiError)
 
 	upsert(
 		ctx context.Context,
 		orgId string,
 		integrationType string,
-		config types.InstalledIntegrationConfig,
-	) (*types.InstalledIntegration, *model.ApiError)
+		config integrationstypes.InstalledIntegrationConfig,
+	) (*integrationstypes.InstalledIntegration, *model.ApiError)
 
 	delete(ctx context.Context, orgId string, integrationType string) *model.ApiError
 }

pkg/query-service/app/integrations/sqlite_repo.go

@@ -7,6 +7,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/query-service/model"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/uptrace/bun"
 )
@@ -26,8 +27,8 @@ func NewInstalledIntegrationsSqliteRepo(store sqlstore.SQLStore) (
 func (r *InstalledIntegrationsSqliteRepo) list(
 	ctx context.Context,
 	orgId string,
-) ([]types.InstalledIntegration, *model.ApiError) {
-	integrations := []types.InstalledIntegration{}
+) ([]integrationstypes.InstalledIntegration, *model.ApiError) {
+	integrations := []integrationstypes.InstalledIntegration{}
 
 	err := r.store.BunDB().NewSelect().
 		Model(&integrations).
@@ -44,8 +45,8 @@ func (r *InstalledIntegrationsSqliteRepo) list(
 
 func (r *InstalledIntegrationsSqliteRepo) get(
 	ctx context.Context, orgId string, integrationTypes []string,
-) (map[string]types.InstalledIntegration, *model.ApiError) {
-	integrations := []types.InstalledIntegration{}
+) (map[string]integrationstypes.InstalledIntegration, *model.ApiError) {
+	integrations := []integrationstypes.InstalledIntegration{}
 
 	typeValues := []interface{}{}
 	for _, integrationType := range integrationTypes {
@@ -62,7 +63,7 @@ func (r *InstalledIntegrationsSqliteRepo) get(
 		))
 	}
 
-	result := map[string]types.InstalledIntegration{}
+	result := map[string]integrationstypes.InstalledIntegration{}
 	for _, ii := range integrations {
 		result[ii.Type] = ii
 	}
@@ -74,10 +75,10 @@ func (r *InstalledIntegrationsSqliteRepo) upsert(
 	ctx context.Context,
 	orgId string,
 	integrationType string,
-	config types.InstalledIntegrationConfig,
-) (*types.InstalledIntegration, *model.ApiError) {
+	config integrationstypes.InstalledIntegrationConfig,
+) (*integrationstypes.InstalledIntegration, *model.ApiError) {
 
-	integration := types.InstalledIntegration{
+	integration := integrationstypes.InstalledIntegration{
 		Identifiable: types.Identifiable{
 			ID: valuer.GenerateUUID(),
 		},
@@ -114,7 +115,7 @@ func (r *InstalledIntegrationsSqliteRepo) delete(
 	ctx context.Context, orgId string, integrationType string,
 ) *model.ApiError {
 	_, dbErr := r.store.BunDB().NewDelete().
-		Model(&types.InstalledIntegration{}).
+		Model(&integrationstypes.InstalledIntegration{}).
 		Where("type = ?", integrationType).
 		Where("org_id = ?", orgId).
 		Exec(ctx)

pkg/query-service/app/server.go

@@ -72,10 +72,7 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 		return nil, err
 	}
 
-	cloudIntegrationsController, err := cloudintegrations.NewController(signoz.SQLStore)
-	if err != nil {
-		return nil, err
-	}
+	cloudIntegrationsRegistry := cloudintegrations.NewCloudProviderRegistry(signoz.Instrumentation.Logger(), signoz.SQLStore, signoz.Querier)
 
 	cacheForTraceDetail, err := memorycache.New(context.TODO(), signoz.Instrumentation.ToProviderSettings(), cache.Config{
 		Provider: "memory",
@@ -128,7 +125,7 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 		Reader:                        reader,
 		RuleManager:                   rm,
 		IntegrationsController:        integrationsController,
-		CloudIntegrationsController:   cloudIntegrationsController,
+		CloudIntegrationsRegistry:     cloudIntegrationsRegistry,
 		LogsParsingPipelineController: logParsingPipelineController,
 		FluxInterval:                  config.Querier.FluxInterval,
 		AlertmanagerAPI:               alertmanager.NewAPI(signoz.Alertmanager),

pkg/types/integration.go

@@ -1,244 +0,0 @@
-package types
-
-import (
-	"database/sql/driver"
-	"encoding/json"
-	"time"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/uptrace/bun"
-)
-
-type IntegrationUserEmail string
-
-const (
-	AWSIntegrationUserEmail IntegrationUserEmail = "aws-integration@signoz.io"
-)
-
-var AllIntegrationUserEmails = []IntegrationUserEmail{
-	AWSIntegrationUserEmail,
-}
-
-// --------------------------------------------------------------------------
-// Normal integration uses just the installed_integration table
-// --------------------------------------------------------------------------
-
-type InstalledIntegration struct {
-	bun.BaseModel `bun:"table:installed_integration"`
-
-	Identifiable
-	Type        string                     `json:"type" bun:"type,type:text,unique:org_id_type"`
-	Config      InstalledIntegrationConfig `json:"config" bun:"config,type:text"`
-	InstalledAt time.Time                  `json:"installed_at" bun:"installed_at,default:current_timestamp"`
-	OrgID       string                     `json:"org_id" bun:"org_id,type:text,unique:org_id_type,references:organizations(id),on_delete:cascade"`
-}
-
-type InstalledIntegrationConfig map[string]interface{}
-
-// For serializing from db
-func (c *InstalledIntegrationConfig) Scan(src interface{}) error {
-	var data []byte
-	switch v := src.(type) {
-	case []byte:
-		data = v
-	case string:
-		data = []byte(v)
-	default:
-		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
-	}
-
-	return json.Unmarshal(data, c)
-}
-
-// For serializing to db
-func (c *InstalledIntegrationConfig) Value() (driver.Value, error) {
-	filterSetJson, err := json.Marshal(c)
-	if err != nil {
-		return nil, errors.WrapInternalf(err, errors.CodeInternal, "could not serialize integration config to JSON")
-	}
-	return filterSetJson, nil
-}
-
-// --------------------------------------------------------------------------
-// Cloud integration uses the cloud_integration table
-// and cloud_integrations_service table
-// --------------------------------------------------------------------------
-
-type CloudIntegration struct {
-	bun.BaseModel `bun:"table:cloud_integration"`
-
-	Identifiable
-	TimeAuditable
-	Provider        string         `json:"provider" bun:"provider,type:text,unique:provider_id"`
-	Config          *AccountConfig `json:"config" bun:"config,type:text"`
-	AccountID       *string        `json:"account_id" bun:"account_id,type:text"`
-	LastAgentReport *AgentReport   `json:"last_agent_report" bun:"last_agent_report,type:text"`
-	RemovedAt       *time.Time     `json:"removed_at" bun:"removed_at,type:timestamp,nullzero"`
-	OrgID           string         `bun:"org_id,type:text,unique:provider_id"`
-}
-
-func (a *CloudIntegration) Status() AccountStatus {
-	status := AccountStatus{}
-	if a.LastAgentReport != nil {
-		lastHeartbeat := a.LastAgentReport.TimestampMillis
-		status.Integration.LastHeartbeatTsMillis = &lastHeartbeat
-	}
-	return status
-}
-
-func (a *CloudIntegration) Account() Account {
-	ca := Account{Id: a.ID.StringValue(), Status: a.Status()}
-
-	if a.AccountID != nil {
-		ca.CloudAccountId = *a.AccountID
-	}
-
-	if a.Config != nil {
-		ca.Config = *a.Config
-	} else {
-		ca.Config = DefaultAccountConfig()
-	}
-	return ca
-}
-
-type Account struct {
-	Id             string        `json:"id"`
-	CloudAccountId string        `json:"cloud_account_id"`
-	Config         AccountConfig `json:"config"`
-	Status         AccountStatus `json:"status"`
-}
-
-type AccountStatus struct {
-	Integration AccountIntegrationStatus `json:"integration"`
-}
-
-type AccountIntegrationStatus struct {
-	LastHeartbeatTsMillis *int64 `json:"last_heartbeat_ts_ms"`
-}
-
-func DefaultAccountConfig() AccountConfig {
-	return AccountConfig{
-		EnabledRegions: []string{},
-	}
-}
-
-type AccountConfig struct {
-	EnabledRegions []string `json:"regions"`
-}
-
-// For serializing from db
-func (c *AccountConfig) Scan(src any) error {
-	var data []byte
-	switch v := src.(type) {
-	case []byte:
-		data = v
-	case string:
-		data = []byte(v)
-	default:
-		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
-	}
-
-	return json.Unmarshal(data, c)
-}
-
-// For serializing to db
-func (c *AccountConfig) Value() (driver.Value, error) {
-	if c == nil {
-		return nil, errors.NewInternalf(errors.CodeInternal, "cloud account config is nil")
-	}
-
-	serialized, err := json.Marshal(c)
-	if err != nil {
-		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't serialize cloud account config to JSON")
-	}
-	return serialized, nil
-}
-
-type AgentReport struct {
-	TimestampMillis int64          `json:"timestamp_millis"`
-	Data            map[string]any `json:"data"`
-}
-
-// For serializing from db
-func (r *AgentReport) Scan(src any) error {
-	var data []byte
-	switch v := src.(type) {
-	case []byte:
-		data = v
-	case string:
-		data = []byte(v)
-	default:
-		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
-	}
-
-	return json.Unmarshal(data, r)
-}
-
-// For serializing to db
-func (r *AgentReport) Value() (driver.Value, error) {
-	if r == nil {
-		return nil, errors.NewInternalf(errors.CodeInternal, "agent report is nil")
-	}
-
-	serialized, err := json.Marshal(r)
-	if err != nil {
-		return nil, errors.WrapInternalf(
-			err, errors.CodeInternal, "couldn't serialize agent report to JSON",
-		)
-	}
-	return serialized, nil
-}
-
-type CloudIntegrationService struct {
-	bun.BaseModel `bun:"table:cloud_integration_service,alias:cis"`
-
-	Identifiable
-	TimeAuditable
-	Type               string             `bun:"type,type:text,notnull,unique:cloud_integration_id_type"`
-	Config             CloudServiceConfig `bun:"config,type:text"`
-	CloudIntegrationID string             `bun:"cloud_integration_id,type:text,notnull,unique:cloud_integration_id_type,references:cloud_integrations(id),on_delete:cascade"`
-}
-
-type CloudServiceLogsConfig struct {
-	Enabled   bool                `json:"enabled"`
-	S3Buckets map[string][]string `json:"s3_buckets,omitempty"`
-}
-
-type CloudServiceMetricsConfig struct {
-	Enabled bool `json:"enabled"`
-}
-
-type CloudServiceConfig struct {
-	Logs    *CloudServiceLogsConfig    `json:"logs,omitempty"`
-	Metrics *CloudServiceMetricsConfig `json:"metrics,omitempty"`
-}
-
-// For serializing from db
-func (c *CloudServiceConfig) Scan(src any) error {
-	var data []byte
-	switch src := src.(type) {
-	case []byte:
-		data = src
-	case string:
-		data = []byte(src)
-	default:
-		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
-	}
-
-	return json.Unmarshal(data, c)
-}
-
-// For serializing to db
-func (c *CloudServiceConfig) Value() (driver.Value, error) {
-	if c == nil {
-		return nil, errors.NewInternalf(errors.CodeInternal, "cloud service config is nil")
-	}
-
-	serialized, err := json.Marshal(c)
-	if err != nil {
-		return nil, errors.WrapInternalf(
-			err, errors.CodeInternal, "couldn't serialize cloud service config to JSON",
-		)
-	}
-	return serialized, nil
-}

pkg/types/integrationstypes/cloudintegration.go

@@ -0,0 +1,723 @@
+package integrationstypes
+
+import (
+	"context"
+	"database/sql/driver"
+	"encoding/json"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/uptrace/bun"
+)
+
+// CloudProvider defines the interface to be implemented by different cloud providers.
+// This is generic interface so it will be accepting and returning generic types instead of concrete.
+// It's the cloud provider's responsibility to cast them to appropriate types and validate
+type CloudProvider interface {
+	GetName() CloudProviderType
+
+	AgentCheckIn(ctx context.Context, req *PostableAgentCheckInPayload) (any, error)
+	GenerateConnectionArtifact(ctx context.Context, req *PostableConnectionArtifact) (any, error)
+	GetAccountStatus(ctx context.Context, orgID, accountID string) (*GettableAccountStatus, error)
+
+	ListServices(ctx context.Context, orgID string, accountID *string) (any, error) // returns either GettableAWSServices or GettableAzureServices
+	GetServiceDetails(ctx context.Context, req *GetServiceDetailsReq) (any, error)
+	ListConnectedAccounts(ctx context.Context, orgID string) (*GettableConnectedAccountsList, error)
+	GetDashboard(ctx context.Context, req *GettableDashboard) (*dashboardtypes.Dashboard, error)
+	GetAvailableDashboards(ctx context.Context, orgID valuer.UUID) ([]*dashboardtypes.Dashboard, error)
+
+	UpdateAccountConfig(ctx context.Context, req *PatchableAccountConfig) (any, error) // req can be either PatchableAWSAccountConfig or PatchableAzureAccountConfig
+	UpdateServiceConfig(ctx context.Context, req *PatchableServiceConfig) (any, error)
+
+	DisconnectAccount(ctx context.Context, orgID, accountID string) (*CloudIntegration, error)
+}
+
+type GettableDashboard struct {
+	ID    string
+	OrgID valuer.UUID
+}
+
+type GettableCloudIntegrationConnectionParams struct {
+	IngestionUrl string `json:"ingestion_url,omitempty"`
+	IngestionKey string `json:"ingestion_key,omitempty"`
+	SigNozAPIUrl string `json:"signoz_api_url,omitempty"`
+	SigNozAPIKey string `json:"signoz_api_key,omitempty"`
+}
+
+type GettableIngestionKey struct {
+	Name  string `json:"name"`
+	Value string `json:"value"`
+	// other attributes from gateway response not included here since they are not being used.
+}
+
+type GettableIngestionKeysSearch struct {
+	Status string                 `json:"status"`
+	Data   []GettableIngestionKey `json:"data"`
+	Error  string                 `json:"error"`
+}
+
+type GettableCreateIngestionKey struct {
+	Status string               `json:"status"`
+	Data   GettableIngestionKey `json:"data"`
+	Error  string               `json:"error"`
+}
+
+type GettableDeployment struct {
+	Name        string `json:"name"`
+	ClusterInfo struct {
+		Region struct {
+			DNS string `json:"dns"`
+		} `json:"region"`
+	} `json:"cluster"`
+}
+
+type GettableConnectedAccountsList struct {
+	Accounts []*Account `json:"accounts"`
+}
+
+// SigNozAWSAgentConfig represents requirements for agent deployment in user's AWS account
+type SigNozAWSAgentConfig struct {
+	// The region in which SigNoz agent should be installed.
+	Region string `json:"region"`
+
+	IngestionUrl string `json:"ingestion_url"`
+	IngestionKey string `json:"ingestion_key"`
+	SigNozAPIUrl string `json:"signoz_api_url"`
+	SigNozAPIKey string `json:"signoz_api_key"`
+
+	Version string `json:"version,omitempty"`
+}
+
+type PostableConnectionArtifact struct {
+	OrgID string
+	Data  []byte // either PostableAWSConnectionUrl or PostableAzureConnectionCommand
+}
+
+type PostableAWSConnectionUrl struct {
+	// Optional. To be specified for updates.
+	// TODO: evaluate and remove if not needed.
+	AccountId     *string               `json:"account_id,omitempty"`
+	AccountConfig *AWSAccountConfig     `json:"account_config"`
+	AgentConfig   *SigNozAWSAgentConfig `json:"agent_config"`
+}
+
+type PostableAzureConnectionCommand struct {
+	AgentConfig   *SigNozAzureAgentConfig `json:"agent_config"`
+	AccountConfig *AzureAccountConfig     `json:"account_config"`
+}
+
+func (p *PostableAWSConnectionUrl) Unmarshal(src any) error {
+	var data []byte
+	switch src := src.(type) {
+	case []byte:
+		data = src
+	case string:
+		data = []byte(src)
+	default:
+		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
+	}
+
+	err := json.Unmarshal(data, p)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize aws connection url request from JSON",
+		)
+	}
+
+	return nil
+}
+
+func (p *PostableAzureConnectionCommand) Unmarshal(src any) error {
+	var data []byte
+	switch src := src.(type) {
+	case []byte:
+		data = src
+	case string:
+		data = []byte(src)
+	default:
+		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
+	}
+
+	err := json.Unmarshal(data, p)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize azure connection command request from JSON",
+		)
+	}
+
+	return nil
+}
+
+type SigNozAzureAgentConfig struct {
+	IngestionUrl string `json:"ingestion_url"`
+	IngestionKey string `json:"ingestion_key"`
+	SigNozAPIUrl string `json:"signoz_api_url"`
+	SigNozAPIKey string `json:"signoz_api_key"`
+
+	Version string `json:"version,omitempty"`
+}
+
+type GettableAWSConnectionUrl struct {
+	AccountId     string `json:"account_id"`
+	ConnectionUrl string `json:"connection_url"`
+}
+
+type GettableAzureConnectionCommand struct {
+	AccountId                   string `json:"account_id"`
+	AzureShellConnectionCommand string `json:"az_shell_connection_command"`
+	AzureCliConnectionCommand   string `json:"az_cli_connection_command"`
+}
+
+type GettableAccountStatus struct {
+	Id             string        `json:"id"`
+	CloudAccountId *string       `json:"cloud_account_id,omitempty"`
+	Status         AccountStatus `json:"status"`
+}
+
+type PostableAgentCheckInPayload struct {
+	ID        string `json:"account_id"`
+	AccountID string `json:"cloud_account_id"`
+	// Arbitrary cloud specific Agent data
+	Data  map[string]any `json:"data,omitempty"`
+	OrgID string         `json:"-"`
+}
+
+type GettableAWSAgentCheckIn struct {
+	AccountId      string     `json:"account_id"`
+	CloudAccountId string     `json:"cloud_account_id"`
+	RemovedAt      *time.Time `json:"removed_at"`
+
+	IntegrationConfig AWSAgentIntegrationConfig `json:"integration_config"`
+}
+
+type AWSAgentIntegrationConfig struct {
+	EnabledRegions              []string               `json:"enabled_regions"`
+	TelemetryCollectionStrategy *AWSCollectionStrategy `json:"telemetry,omitempty"`
+}
+
+type AzureAgentIntegrationConfig struct {
+	DeploymentRegion            string                   `json:"deployment_region"` // will not be changed once set
+	EnabledResourceGroups       []string                 `json:"resource_groups"`
+	TelemetryCollectionStrategy *AzureCollectionStrategy `json:"telemetry,omitempty"`
+}
+
+type GettableAzureAgentCheckIn struct {
+	AccountId         string                      `json:"account_id"`
+	CloudAccountId    string                      `json:"cloud_account_id"`
+	RemovedAt         *time.Time                  `json:"removed_at"`
+	IntegrationConfig AzureAgentIntegrationConfig `json:"integration_config"`
+}
+
+type PatchableServiceConfig struct {
+	OrgID     string `json:"org_id"`
+	ServiceId string `json:"service_id"`
+	Config    []byte `json:"config"` // json serialized config
+}
+
+type PatchableAWSCloudServiceConfig struct {
+	CloudAccountId string                 `json:"cloud_account_id"`
+	Config         *AWSCloudServiceConfig `json:"config"`
+}
+
+type AWSCloudServiceConfig struct {
+	Logs    *AWSCloudServiceLogsConfig    `json:"logs,omitempty"`
+	Metrics *AWSCloudServiceMetricsConfig `json:"metrics,omitempty"`
+}
+
+// Unmarshal unmarshalls data from src
+func (c *PatchableAWSCloudServiceConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, c)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize aws service config req from JSON",
+		)
+	}
+
+	return nil
+}
+
+// Marshal serializes data to bytes
+func (c *PatchableAWSCloudServiceConfig) Marshal() ([]byte, error) {
+	serialized, err := json.Marshal(c)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize aws service config req to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+// Unmarshal unmarshalls data from src
+func (a *AWSCloudServiceConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, a)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize cloud service config from JSON",
+		)
+	}
+
+	return nil
+}
+
+// Marshal serializes data to bytes
+func (a *AWSCloudServiceConfig) Marshal() ([]byte, error) {
+	serialized, err := json.Marshal(a)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize cloud service config to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+func (a *AWSCloudServiceConfig) Validate(def *AWSServiceDefinition) error {
+	if def.Id != S3Sync && a.Logs != nil && a.Logs.S3Buckets != nil {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "s3 buckets can only be added to service-type[%s]", S3Sync)
+	} else if def.Id == S3Sync && a.Logs != nil && a.Logs.S3Buckets != nil {
+		for region := range a.Logs.S3Buckets {
+			if _, found := ValidAWSRegions[region]; !found {
+				return errors.NewInvalidInputf(CodeInvalidCloudRegion, "invalid cloud region: %s", region)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (a *AzureCloudServiceConfig) Validate(def *AzureServiceDefinition) error {
+	logsMap := make(map[string]bool)
+	metricsMap := make(map[string]bool)
+
+	for _, log := range def.Strategy.AzureLogs {
+		logsMap[log.Name] = true
+	}
+
+	for _, metric := range def.Strategy.AzureMetrics {
+		metricsMap[metric.Name] = true
+	}
+
+	for _, log := range a.Logs {
+		if _, found := logsMap[log.Name]; !found {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid log name: %s", log.Name)
+		}
+	}
+
+	for _, metric := range a.Metrics {
+		if _, found := metricsMap[metric.Name]; !found {
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid metric name: %s", metric.Name)
+		}
+	}
+
+	return nil
+}
+
+type AzureCloudServiceLogsConfig struct {
+	Enabled bool   `json:"enabled"`
+	Name    string `json:"name"`
+}
+
+type AzureCloudServiceMetricsConfig struct {
+	Enabled bool   `json:"enabled"`
+	Name    string `json:"name"`
+}
+
+type PatchServiceConfigResponse struct {
+	ServiceId string `json:"id"`
+	Config    any    `json:"config"`
+}
+
+type PatchableAccountConfig struct {
+	OrgID     string
+	AccountId string
+	Data      []byte // can be either AWSAccountConfig or AzureAccountConfig
+}
+
+type PatchableAWSAccountConfig struct {
+	Config *AWSAccountConfig `json:"config"`
+}
+
+type PatchableAzureAccountConfig struct {
+	Config *AzureAccountConfig `json:"config"`
+}
+
+func (p *PatchableAWSAccountConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, p)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize patchable account config from JSON",
+		)
+	}
+
+	return nil
+}
+
+func (p *PatchableAWSAccountConfig) Marshal() ([]byte, error) {
+	if p == nil {
+		return nil, errors.NewInternalf(errors.CodeInternal, "patchable account config is nil")
+	}
+
+	serialized, err := json.Marshal(p)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize patchable account config to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+func (p *PatchableAzureAccountConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, p)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize patchable account config from JSON",
+		)
+	}
+
+	return nil
+}
+
+func (p *PatchableAzureAccountConfig) Marshal() ([]byte, error) {
+	if p == nil {
+		return nil, errors.NewInternalf(errors.CodeInternal, "patchable account config is nil")
+	}
+
+	serialized, err := json.Marshal(p)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize patchable account config to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+type AWSAccountConfig struct {
+	EnabledRegions []string `json:"regions"`
+}
+
+type AzureAccountConfig struct {
+	DeploymentRegion      string   `json:"deployment_region,omitempty"`
+	EnabledResourceGroups []string `json:"resource_groups,omitempty"`
+}
+
+// Unmarshal unmarshalls data from src
+func (c *AWSAccountConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, c)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize AWS account config from JSON",
+		)
+	}
+
+	return nil
+}
+
+// Marshal serializes data to bytes
+func (c *AWSAccountConfig) Marshal() ([]byte, error) {
+	if c == nil {
+		return nil, errors.NewInternalf(errors.CodeInternal, "cloud account config is nil")
+	}
+
+	serialized, err := json.Marshal(c)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't serialize cloud account config to JSON")
+	}
+	return serialized, nil
+}
+
+// Unmarshal unmarshalls data from src
+func (c *AzureAccountConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, c)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize azure account config from JSON",
+		)
+	}
+
+	return nil
+}
+
+// Marshal serializes data to bytes
+func (c *AzureAccountConfig) Marshal() ([]byte, error) {
+	if c == nil {
+		return nil, errors.NewInternalf(errors.CodeInternal, "cloud account config is nil")
+	}
+
+	serialized, err := json.Marshal(c)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "couldn't serialize cloud account config to JSON")
+	}
+	return serialized, nil
+}
+
+type GettableAWSServices struct {
+	Services []AWSServiceSummary `json:"services"`
+}
+
+type GettableAzureServices struct {
+	Services []AzureServiceSummary `json:"services"`
+}
+
+type PatchableAzureCloudServiceConfig struct {
+	CloudAccountId string                   `json:"cloud_account_id"`
+	Config         *AzureCloudServiceConfig `json:"config"`
+}
+
+type AzureCloudServiceConfig struct {
+	Logs    []*AzureCloudServiceLogsConfig    `json:"logs,omitempty"`
+	Metrics []*AzureCloudServiceMetricsConfig `json:"metrics,omitempty"`
+}
+
+// Unmarshal unmarshalls data from src
+func (c *PatchableAzureCloudServiceConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, c)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize cloud service config from JSON",
+		)
+	}
+
+	return nil
+}
+
+// Marshal serializes data to bytes
+func (c *PatchableAzureCloudServiceConfig) Marshal() ([]byte, error) {
+	if c == nil {
+		return nil, errors.NewInternalf(errors.CodeInternal, "cloud service config is nil")
+	}
+
+	serialized, err := json.Marshal(c)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize cloud service config to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+func (a *AzureCloudServiceConfig) Unmarshal(src []byte) error {
+	err := json.Unmarshal(src, a)
+	if err != nil {
+		return errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't deserialize cloud service config from JSON",
+		)
+	}
+
+	return nil
+}
+
+func (a *AzureCloudServiceConfig) Marshal() ([]byte, error) {
+	serialized, err := json.Marshal(a)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize cloud service config to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+type GetServiceDetailsReq struct {
+	OrgID          string
+	ServiceId      string
+	CloudAccountID *string
+}
+
+// --------------------------------------------------------------------------
+// DATABASE TYPES
+// --------------------------------------------------------------------------
+
+// --------------------------------------------------------------------------
+// Cloud integration uses the cloud_integration table
+// and cloud_integrations_service table
+// --------------------------------------------------------------------------
+
+type CloudIntegration struct {
+	bun.BaseModel `bun:"table:cloud_integration"`
+
+	types.Identifiable
+	types.TimeAuditable
+	Provider        string       `json:"provider" bun:"provider,type:text,unique:provider_id"`
+	Config          []byte       `json:"config" bun:"config,type:text"` // json serialized config
+	AccountID       *string      `json:"account_id" bun:"account_id,type:text"`
+	LastAgentReport *AgentReport `json:"last_agent_report" bun:"last_agent_report,type:text"`
+	RemovedAt       *time.Time   `json:"removed_at" bun:"removed_at,type:timestamp,nullzero"`
+	OrgID           string       `bun:"org_id,type:text,unique:provider_id"`
+}
+
+func (a *CloudIntegration) Status() AccountStatus {
+	status := AccountStatus{}
+	if a.LastAgentReport != nil {
+		lastHeartbeat := a.LastAgentReport.TimestampMillis
+		status.Integration.LastHeartbeatTsMillis = &lastHeartbeat
+	}
+	return status
+}
+
+func (a *CloudIntegration) Account(cloudProvider CloudProviderType) *Account {
+	ca := &Account{Id: a.ID.StringValue(), Status: a.Status()}
+
+	if a.AccountID != nil {
+		ca.CloudAccountId = *a.AccountID
+	}
+
+	ca.Config = map[string]interface{}{}
+
+	if len(a.Config) < 1 {
+		return ca
+	}
+
+	switch cloudProvider {
+	case CloudProviderAWS:
+		config := new(AWSAccountConfig)
+		_ = config.Unmarshal(a.Config)
+		ca.Config = config
+	case CloudProviderAzure:
+		config := new(AzureAccountConfig)
+		_ = config.Unmarshal(a.Config)
+		ca.Config = config
+	default:
+	}
+
+	return ca
+}
+
+type Account struct {
+	Id             string        `json:"id"`
+	CloudAccountId string        `json:"cloud_account_id"`
+	Config         any           `json:"config"` // AWSAccountConfig or AzureAccountConfig
+	Status         AccountStatus `json:"status"`
+}
+
+type AccountStatus struct {
+	Integration AccountIntegrationStatus `json:"integration"`
+}
+
+type AccountIntegrationStatus struct {
+	LastHeartbeatTsMillis *int64 `json:"last_heartbeat_ts_ms"`
+}
+
+func DefaultAWSAccountConfig() AWSAccountConfig {
+	return AWSAccountConfig{
+		EnabledRegions: []string{},
+	}
+}
+
+func DefaultAzureAccountConfig() AzureAccountConfig {
+	return AzureAccountConfig{
+		DeploymentRegion:      "",
+		EnabledResourceGroups: []string{},
+	}
+}
+
+type AWSServiceSummary struct {
+	DefinitionMetadata
+	Config *AWSCloudServiceConfig `json:"config"`
+}
+
+type AzureServiceSummary struct {
+	DefinitionMetadata
+	Config *AzureCloudServiceConfig `json:"config"`
+}
+
+type GettableAWSServiceDetails struct {
+	AWSServiceDefinition
+	Config           *AWSCloudServiceConfig   `json:"config"`
+	ConnectionStatus *ServiceConnectionStatus `json:"status,omitempty"`
+}
+
+type GettableAzureServiceDetails struct {
+	AzureServiceDefinition
+	Config           *AzureCloudServiceConfig `json:"config"`
+	ConnectionStatus *ServiceConnectionStatus `json:"status,omitempty"`
+}
+
+type ServiceConnectionStatus struct {
+	Logs    []*SignalConnectionStatus `json:"logs"`
+	Metrics []*SignalConnectionStatus `json:"metrics"`
+}
+
+type SignalConnectionStatus struct {
+	CategoryID           string `json:"category"`
+	CategoryDisplayName  string `json:"category_display_name"`
+	LastReceivedTsMillis int64  `json:"last_received_ts_ms"` // epoch milliseconds
+	LastReceivedFrom     string `json:"last_received_from"`  // resource identifier
+}
+
+//// AddAWSServiceStrategy is a helper for accumulating strategies for enabled services.
+//func AddAWSServiceStrategy(serviceType string, cs *AWSCollectionStrategy,
+//	definitionStrat *AWSCollectionStrategy, config *AWSCloudServiceConfig) {
+//	if config.Logs != nil && config.Logs.Enabled {
+//		if serviceType == S3Sync {
+//			// S3 bucket sync; No cloudwatch logs are appended for this service type;
+//			// Though definition is populated with a custom cloudwatch group that helps in calculating logs connection status
+//			cs.S3Buckets = config.Logs.S3Buckets
+//		} else if definitionStrat.AWSLogs != nil { // services that includes a logs subscription
+//			cs.AWSLogs.Subscriptions = append(
+//				cs.AWSLogs.Subscriptions,
+//				definitionStrat.AWSLogs.Subscriptions...,
+//			)
+//		}
+//	}
+//	if config.Metrics != nil && config.Metrics.Enabled && definitionStrat.AWSMetrics != nil {
+//		cs.AWSMetrics.StreamFilters = append(
+//			cs.AWSMetrics.StreamFilters,
+//			definitionStrat.AWSMetrics.StreamFilters...,
+//		)
+//	}
+//}
+
+type AgentReport struct {
+	TimestampMillis int64          `json:"timestamp_millis"`
+	Data            map[string]any `json:"data"`
+}
+
+// Scan scans data from db
+func (r *AgentReport) Scan(src any) error {
+	var data []byte
+	switch v := src.(type) {
+	case []byte:
+		data = v
+	case string:
+		data = []byte(v)
+	default:
+		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
+	}
+
+	return json.Unmarshal(data, r)
+}
+
+// Value serializes data to bytes for db insertion
+func (r *AgentReport) Value() (driver.Value, error) {
+	if r == nil {
+		return nil, errors.NewInternalf(errors.CodeInternal, "agent report is nil")
+	}
+
+	serialized, err := json.Marshal(r)
+	if err != nil {
+		return nil, errors.WrapInternalf(
+			err, errors.CodeInternal, "couldn't serialize agent report to JSON",
+		)
+	}
+	return serialized, nil
+}
+
+type CloudIntegrationService struct {
+	bun.BaseModel `bun:"table:cloud_integration_service,alias:cis"`
+
+	types.Identifiable
+	types.TimeAuditable
+	Type               string `bun:"type,type:text,notnull,unique:cloud_integration_id_type"`
+	Config             []byte `bun:"config,type:text"` // json serialized config
+	CloudIntegrationID string `bun:"cloud_integration_id,type:text,notnull,unique:cloud_integration_id_type,references:cloud_integrations(id),on_delete:cascade"`
+}
+
+type AWSCloudServiceLogsConfig struct {
+	Enabled   bool                `json:"enabled"`
+	S3Buckets map[string][]string `json:"s3_buckets,omitempty"`
+}
+
+type AWSCloudServiceMetricsConfig struct {
+	Enabled bool `json:"enabled"`
+}

pkg/types/integrationstypes/cloudservicedefinitions.go

@@ -0,0 +1,249 @@
+package integrationstypes
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+const (
+	S3Sync = "s3sync"
+)
+
+type DefinitionMetadata struct {
+	Id    string `json:"id"`
+	Title string `json:"title"`
+	Icon  string `json:"icon"`
+}
+
+type Definition interface {
+	GetId() string
+	Validate() error
+	PopulateDashboardURLs(svcId string)
+}
+
+var _ Definition = &AWSServiceDefinition{}
+var _ Definition = &AzureServiceDefinition{}
+
+type AWSServiceDefinition struct {
+	DefinitionMetadata
+	Overview             string                 `json:"overview"` // markdown
+	Assets               Assets                 `json:"assets"`
+	SupportedSignals     SupportedSignals       `json:"supported_signals"`
+	DataCollected        DataCollected          `json:"data_collected"`
+	Strategy             *AWSCollectionStrategy `json:"telemetry_collection_strategy"`
+	IngestionStatusCheck *IngestionStatusCheck  `json:"ingestion_status_check"`
+}
+
+type IngestionStatusCheck struct {
+	Metrics []*IngestionStatusCheckCategory `json:"metrics"`
+	Logs    []*IngestionStatusCheckCategory `json:"logs"`
+}
+
+type IngestionStatusCheckCategory struct {
+	Category    string                           `json:"category"`
+	DisplayName string                           `json:"display_name"`
+	Checks      []*IngestionStatusCheckAttribute `json:"checks"`
+}
+
+type IngestionStatusCheckAttribute struct {
+	Key        string                                 `json:"key"` // search key (metric name or log message)
+	Attributes []*IngestionStatusCheckAttributeFilter `json:"attributes"`
+}
+
+type IngestionStatusCheckAttributeFilter struct {
+	Name     string `json:"name"`
+	Operator string `json:"operator"`
+	Value    string `json:"value"`
+}
+
+func (def *AWSServiceDefinition) GetId() string {
+	return def.Id
+}
+
+func (def *AWSServiceDefinition) Validate() error {
+	seenDashboardIds := map[string]interface{}{}
+
+	if def.Strategy == nil {
+		return errors.NewInternalf(errors.CodeInternal, "telemetry_collection_strategy is required")
+	}
+
+	for _, dd := range def.Assets.Dashboards {
+		if _, seen := seenDashboardIds[dd.Id]; seen {
+			return errors.NewInternalf(errors.CodeInternal, "multiple dashboards found with id %s for AWS Integration", dd.Id)
+		}
+		seenDashboardIds[dd.Id] = nil
+	}
+
+	return nil
+}
+
+func (def *AWSServiceDefinition) PopulateDashboardURLs(serviceId string) {
+	for i := range def.Assets.Dashboards {
+		dashboardId := def.Assets.Dashboards[i].Id
+		url := "/dashboard/" + GetCloudIntegrationDashboardID(CloudProviderAWS, serviceId, dashboardId)
+		def.Assets.Dashboards[i].Url = url
+	}
+}
+
+type AzureServiceDefinition struct {
+	DefinitionMetadata
+
+	Overview string `json:"overview"` // markdown
+
+	Assets Assets `json:"assets"`
+
+	SupportedSignals SupportedSignals `json:"supported_signals"`
+
+	DataCollected DataCollected `json:"data_collected"`
+
+	Strategy *AzureCollectionStrategy `json:"telemetry_collection_strategy"`
+}
+
+func (def *AzureServiceDefinition) PopulateDashboardURLs(svcId string) {
+	for i := range def.Assets.Dashboards {
+		dashboardId := def.Assets.Dashboards[i].Id
+		url := "/dashboard/" + GetCloudIntegrationDashboardID(CloudProviderAzure, svcId, dashboardId)
+		def.Assets.Dashboards[i].Url = url
+	}
+}
+
+func (def *AzureServiceDefinition) GetId() string {
+	return def.Id
+}
+
+func (def *AzureServiceDefinition) Validate() error {
+	seenDashboardIds := map[string]interface{}{}
+
+	if def.Strategy == nil {
+		return errors.NewInternalf(errors.CodeInternal, "telemetry_collection_strategy is required")
+	}
+
+	for _, dd := range def.Assets.Dashboards {
+		if _, seen := seenDashboardIds[dd.Id]; seen {
+			return errors.NewInternalf(errors.CodeInternal, "multiple dashboards found with id %s for Azure Integration", dd.Id)
+		}
+		seenDashboardIds[dd.Id] = nil
+	}
+
+	return nil
+}
+
+type Assets struct {
+	Dashboards []Dashboard `json:"dashboards"`
+}
+
+type SupportedSignals struct {
+	Logs    bool `json:"logs"`
+	Metrics bool `json:"metrics"`
+}
+
+type DataCollected struct {
+	Logs    []CollectedLogAttribute `json:"logs"`
+	Metrics []CollectedMetric       `json:"metrics"`
+}
+
+type CollectedLogAttribute struct {
+	Name string `json:"name"`
+	Path string `json:"path"`
+	Type string `json:"type"`
+}
+
+type CollectedMetric struct {
+	Name        string `json:"name"`
+	Type        string `json:"type"`
+	Unit        string `json:"unit"`
+	Description string `json:"description"`
+}
+
+type AWSCollectionStrategy struct {
+	Provider valuer.String `json:"provider"`
+
+	AWSMetrics *AWSMetricsStrategy `json:"aws_metrics,omitempty"`
+	AWSLogs    *AWSLogsStrategy    `json:"aws_logs,omitempty"`
+	S3Buckets  map[string][]string `json:"s3_buckets,omitempty"` // Only available in S3 Sync Service Type
+}
+
+type AzureCollectionStrategy struct {
+	Provider valuer.String `json:"provider"`
+
+	AzureMetrics []*AzureMetricsStrategy `json:"azure_metrics"`
+	AzureLogs    []*AzureLogsStrategy    `json:"azure_logs"`
+}
+
+type AzureResourceGroup struct {
+	Name   string `json:"name"`
+	Region string `json:"region"`
+}
+
+type AWSMetricsStrategy struct {
+	// to be used as https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-cloudwatch-metricstream.html#cfn-cloudwatch-metricstream-includefilters
+	StreamFilters []struct {
+		// json tags here are in the shape expected by AWS API as detailed at
+		// https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-cloudwatch-metricstream-metricstreamfilter.html
+		Namespace   string   `json:"Namespace"`
+		MetricNames []string `json:"MetricNames,omitempty"`
+	} `json:"cloudwatch_metric_stream_filters"`
+}
+
+type AWSLogsStrategy struct {
+	Subscriptions []struct {
+		// subscribe to all logs groups with specified prefix.
+		// eg: `/aws/rds/`
+		LogGroupNamePrefix string `json:"log_group_name_prefix"`
+
+		// https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/FilterAndPatternSyntax.html
+		// "" implies no filtering is required.
+		FilterPattern string `json:"filter_pattern"`
+	} `json:"cloudwatch_logs_subscriptions"`
+}
+
+type AzureMetricsStrategy struct {
+	CategoryType string `json:"category_type"`
+	Name         string `json:"name"`
+}
+
+type AzureLogsStrategy struct {
+	CategoryType string `json:"category_type"`
+	Name         string `json:"name"`
+}
+
+type Dashboard struct {
+	Id          string                                `json:"id"`
+	Url         string                                `json:"url"`
+	Title       string                                `json:"title"`
+	Description string                                `json:"description"`
+	Image       string                                `json:"image"`
+	Definition  *dashboardtypes.StorableDashboardData `json:"definition,omitempty"`
+}
+
+func GetCloudIntegrationDashboardID(cloudProvider valuer.String, svcId, dashboardId string) string {
+	return fmt.Sprintf("cloud-integration--%s--%s--%s", cloudProvider, svcId, dashboardId)
+}
+
+func GetDashboardsFromAssets(svcId string, cloudProvider CloudProviderType, createdAt *time.Time, assets Assets) []*dashboardtypes.Dashboard {
+	dashboards := make([]*dashboardtypes.Dashboard, 0)
+
+	for _, d := range assets.Dashboards {
+		author := fmt.Sprintf("%s-integration", cloudProvider)
+		dashboards = append(dashboards, &dashboardtypes.Dashboard{
+			ID:     GetCloudIntegrationDashboardID(cloudProvider, svcId, d.Id),
+			Locked: true,
+			Data:   *d.Definition,
+			TimeAuditable: types.TimeAuditable{
+				CreatedAt: *createdAt,
+				UpdatedAt: *createdAt,
+			},
+			UserAuditable: types.UserAuditable{
+				CreatedBy: author,
+				UpdatedBy: author,
+			},
+		})
+	}
+
+	return dashboards
+}

pkg/types/integrationstypes/constants.go

@@ -1,4 +1,4 @@
-package cloudintegrations
+package integrationstypes
 
 import (
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -41,3 +41,62 @@ var ValidAWSRegions = map[string]bool{
 	"us-west-1":      true, // US West (N. California).
 	"us-west-2":      true, // US West (Oregon).
 }
+
+var ValidAzureRegions = map[string]bool{
+	"australiacentral":   true,
+	"australiacentral2":  true,
+	"australiaeast":      true,
+	"australiasoutheast": true,
+	"austriaeast":        true,
+	"belgiumcentral":     true,
+	"brazilsouth":        true,
+	"brazilsoutheast":    true,
+	"canadacentral":      true,
+	"canadaeast":         true,
+	"centralindia":       true,
+	"centralus":          true,
+	"chilecentral":       true,
+	"denmarkeast":        true,
+	"eastasia":           true,
+	"eastus":             true,
+	"eastus2":            true,
+	"francecentral":      true,
+	"francesouth":        true,
+	"germanynorth":       true,
+	"germanywestcentral": true,
+	"indonesiacentral":   true,
+	"israelcentral":      true,
+	"italynorth":         true,
+	"japaneast":          true,
+	"japanwest":          true,
+	"koreacentral":       true,
+	"koreasouth":         true,
+	"malaysiawest":       true,
+	"mexicocentral":      true,
+	"newzealandnorth":    true,
+	"northcentralus":     true,
+	"northeurope":        true,
+	"norwayeast":         true,
+	"norwaywest":         true,
+	"polandcentral":      true,
+	"qatarcentral":       true,
+	"southafricanorth":   true,
+	"southafricawest":    true,
+	"southcentralus":     true,
+	"southindia":         true,
+	"southeastasia":      true,
+	"spaincentral":       true,
+	"swedencentral":      true,
+	"switzerlandnorth":   true,
+	"switzerlandwest":    true,
+	"uaecentral":         true,
+	"uaenorth":           true,
+	"uksouth":            true,
+	"ukwest":             true,
+	"westcentralus":      true,
+	"westeurope":         true,
+	"westindia":          true,
+	"westus":             true,
+	"westus2":            true,
+	"westus3":            true,
+}

pkg/types/integrationstypes/integration.go

@@ -0,0 +1,109 @@
+package integrationstypes
+
+import (
+	"database/sql/driver"
+	"encoding/json"
+	"strings"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/uptrace/bun"
+)
+
+// CloudProviderType type alias
+type CloudProviderType = valuer.String
+
+var (
+	CloudProviderAWS   = valuer.NewString("aws")
+	CloudProviderAzure = valuer.NewString("azure")
+)
+
+var (
+	CodeCloudProviderInvalidInput = errors.MustNewCode("invalid_cloud_provider")
+)
+
+func NewCloudProvider(provider string) (CloudProviderType, error) {
+	switch provider {
+	case CloudProviderAWS.String(), CloudProviderAzure.String():
+		return valuer.NewString(provider), nil
+	default:
+		return CloudProviderType{}, errors.NewInvalidInputf(CodeCloudProviderInvalidInput, "invalid cloud provider: %s", provider)
+	}
+}
+
+var (
+	AWSIntegrationUserEmail   = valuer.MustNewEmail("aws-integration@signoz.io")
+	AzureIntegrationUserEmail = valuer.MustNewEmail("azure-integration@signoz.io")
+)
+
+var IntegrationUserEmails = []valuer.Email{
+	AWSIntegrationUserEmail,
+	AzureIntegrationUserEmail,
+}
+
+func IsCloudIntegrationDashboardUuid(dashboardUuid string) bool {
+	parts := strings.SplitN(dashboardUuid, "--", 4)
+	if len(parts) != 4 {
+		return false
+	}
+
+	return parts[0] == "cloud-integration"
+}
+
+func GetCloudProviderFromDashboardID(dashboardUuid string) (CloudProviderType, error) {
+	parts := strings.SplitN(dashboardUuid, "--", 4)
+	if len(parts) != 4 {
+		return valuer.String{}, errors.NewInvalidInputf(CodeCloudProviderInvalidInput, "invalid dashboard uuid: %s", dashboardUuid)
+	}
+
+	providerStr := parts[1]
+
+	cloudProvider, err := NewCloudProvider(providerStr)
+	if err != nil {
+		return CloudProviderType{}, err
+	}
+
+	return cloudProvider, nil
+}
+
+// --------------------------------------------------------------------------
+// Normal integration uses just the installed_integration table
+// --------------------------------------------------------------------------
+
+type InstalledIntegration struct {
+	bun.BaseModel `bun:"table:installed_integration"`
+
+	types.Identifiable
+	Type        string                     `json:"type" bun:"type,type:text,unique:org_id_type"`
+	Config      InstalledIntegrationConfig `json:"config" bun:"config,type:text"`
+	InstalledAt time.Time                  `json:"installed_at" bun:"installed_at,default:current_timestamp"`
+	OrgID       string                     `json:"org_id" bun:"org_id,type:text,unique:org_id_type,references:organizations(id),on_delete:cascade"`
+}
+
+type InstalledIntegrationConfig map[string]interface{}
+
+// Scan scans data from db
+func (c *InstalledIntegrationConfig) Scan(src interface{}) error {
+	var data []byte
+	switch v := src.(type) {
+	case []byte:
+		data = v
+	case string:
+		data = []byte(v)
+	default:
+		return errors.NewInternalf(errors.CodeInternal, "tried to scan from %T instead of string or bytes", src)
+	}
+
+	return json.Unmarshal(data, c)
+}
+
+// Value serializes data to db
+func (c *InstalledIntegrationConfig) Value() (driver.Value, error) {
+	filterSetJson, err := json.Marshal(c)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, errors.CodeInternal, "could not serialize integration config to JSON")
+	}
+	return filterSetJson, nil
+}