refactor: using QBv5 for connection status

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

@@ -7,18 +7,18 @@ import (
 	"net/url"
 	"slices"
 	"sort"
-	"strings"
 	"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/interfaces"
-	"github.com/SigNoz/signoz/pkg/query-service/model"
-	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
 	"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"
 )
@@ -30,19 +30,17 @@ var (
 
 type awsProvider struct {
 	logger                *slog.Logger
-	querier               interfaces.Querier
+	querier               querier.Querier
 	accountsRepo          integrationstore.CloudProviderAccountsRepository
 	serviceConfigRepo     integrationstore.ServiceConfigDatabase
 	awsServiceDefinitions *services.AWSServicesProvider
-	reader                interfaces.Reader
 }
 
 func NewAWSCloudProvider(
 	logger *slog.Logger,
 	accountsRepo integrationstore.CloudProviderAccountsRepository,
 	serviceConfigRepo integrationstore.ServiceConfigDatabase,
-	reader interfaces.Reader,
-	querier interfaces.Querier,
+	querier querier.Querier,
 ) integrationstypes.CloudProvider {
 	awsServiceDefinitions, err := services.NewAWSCloudProviderServices()
 	if err != nil {
@@ -51,7 +49,6 @@ func NewAWSCloudProvider(
 
 	return &awsProvider{
 		logger:                logger,
-		reader:                reader,
 		querier:               querier,
 		accountsRepo:          accountsRepo,
 		serviceConfigRepo:     serviceConfigRepo,
@@ -282,11 +279,12 @@ func (a *awsProvider) GetServiceDetails(ctx context.Context, req *integrationsty
 
 	details.Config = config
 
-	connectionStatus, err := a.calculateCloudIntegrationServiceConnectionStatus(
+	connectionStatus, err := a.getServiceConnectionStatus(
 		ctx,
-		req.OrgID,
 		*req.CloudAccountID,
+		req.OrgID,
 		&details.AWSServiceDefinition,
+		config,
 	)
 	if err != nil {
 		return nil, err
@@ -297,204 +295,241 @@ func (a *awsProvider) GetServiceDetails(ctx context.Context, req *integrationsty
 	return details, nil
 }
 
-func (a *awsProvider) calculateCloudIntegrationServiceConnectionStatus(
+func (a *awsProvider) getServiceConnectionStatus(
 	ctx context.Context,
+	cloudAccountID string,
 	orgID valuer.UUID,
-	cloudAccountId string,
-	svcDetails *integrationstypes.AWSServiceDefinition,
+	def *integrationstypes.AWSServiceDefinition,
+	serviceConfig *integrationstypes.AWSCloudServiceConfig,
 ) (*integrationstypes.ServiceConnectionStatus, error) {
-	telemetryCollectionStrategy := svcDetails.Strategy
-	result := &integrationstypes.ServiceConnectionStatus{}
-
-	if telemetryCollectionStrategy == nil {
-		return result, model.InternalError(fmt.Errorf(
-			"service doesn't have telemetry collection strategy: %s", svcDetails.Id,
-		))
+	if def.Strategy == nil {
+		return nil, nil
 	}
 
-	errors := make([]error, 0)
-	var resultLock sync.Mutex
+	resp := new(integrationstypes.ServiceConnectionStatus)
 
-	var wg sync.WaitGroup
+	wg := sync.WaitGroup{}
+	wg.Add(2)
 
-	// Calculate metrics connection status
-	if telemetryCollectionStrategy.AWSMetrics != nil {
-		wg.Add(1)
+	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()
-
-			metricsConnStatus, apiErr := a.calculateAWSIntegrationSvcMetricsConnectionStatus(
-				ctx, cloudAccountId, telemetryCollectionStrategy.AWSMetrics, svcDetails.DataCollected.Metrics,
-			)
-
-			resultLock.Lock()
-			defer resultLock.Unlock()
-
-			if apiErr != nil {
-				errors = append(errors, apiErr)
-			} else {
-				result.Metrics = metricsConnStatus
-			}
+			status, _ := a.getServiceMetricsConnectionStatus(ctx, cloudAccountID, orgID, def)
+			resp.Metrics = status
 		}()
 	}
 
-	// Calculate logs connection status
-	if telemetryCollectionStrategy.AWSLogs != nil {
-		wg.Add(1)
+	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()
-
-			logsConnStatus, apiErr := a.calculateAWSIntegrationSvcLogsConnectionStatus(
-				ctx, orgID, cloudAccountId, telemetryCollectionStrategy.AWSLogs,
-			)
-
-			resultLock.Lock()
-			defer resultLock.Unlock()
-
-			if apiErr != nil {
-				errors = append(errors, apiErr)
-			} else {
-				result.Logs = logsConnStatus
-			}
+			status, _ := a.getServiceLogsConnectionStatus(ctx, cloudAccountID, orgID, def)
+			resp.Logs = status
 		}()
 	}
 
 	wg.Wait()
 
-	if len(errors) > 0 {
-		return result, errors[0]
-	}
-
-	return result, nil
-
-}
-func (a *awsProvider) calculateAWSIntegrationSvcMetricsConnectionStatus(
-	ctx context.Context,
-	cloudAccountId string,
-	strategy *integrationstypes.AWSMetricsStrategy,
-	metricsCollectedBySvc []integrationstypes.CollectedMetric,
-) (*integrationstypes.SignalConnectionStatus, *model.ApiError) {
-	if strategy == nil || len(strategy.StreamFilters) < 1 {
-		return nil, nil
-	}
-
-	expectedLabelValues := map[string]string{
-		"cloud_provider":   "aws",
-		"cloud_account_id": cloudAccountId,
-	}
-
-	metricsNamespace := strategy.StreamFilters[0].Namespace
-	metricsNamespaceParts := strings.Split(metricsNamespace, "/")
-
-	if len(metricsNamespaceParts) >= 2 {
-		expectedLabelValues["service_namespace"] = metricsNamespaceParts[0]
-		expectedLabelValues["service_name"] = metricsNamespaceParts[1]
-	} else {
-		// metrics for single word namespaces like "CWAgent" do not
-		// have the service_namespace label populated
-		expectedLabelValues["service_name"] = metricsNamespaceParts[0]
-	}
-
-	metricNamesCollectedBySvc := []string{}
-	for _, cm := range metricsCollectedBySvc {
-		metricNamesCollectedBySvc = append(metricNamesCollectedBySvc, cm.Name)
-	}
-
-	statusForLastReceivedMetric, apiErr := a.reader.GetLatestReceivedMetric(
-		ctx, metricNamesCollectedBySvc, expectedLabelValues,
-	)
-	if apiErr != nil {
-		return nil, apiErr
-	}
-
-	if statusForLastReceivedMetric != nil {
-		return &integrationstypes.SignalConnectionStatus{
-			LastReceivedTsMillis: statusForLastReceivedMetric.LastReceivedTsMillis,
-			LastReceivedFrom:     "signoz-aws-integration",
-		}, nil
-	}
-
-	return nil, nil
+	return resp, nil
 }
 
-func (a *awsProvider) calculateAWSIntegrationSvcLogsConnectionStatus(
+func (a *awsProvider) getServiceMetricsConnectionStatus(
 	ctx context.Context,
+	cloudAccountID string,
 	orgID valuer.UUID,
-	cloudAccountId string,
-	strategy *integrationstypes.AWSLogsStrategy,
-) (*integrationstypes.SignalConnectionStatus, *model.ApiError) {
-	if strategy == nil || len(strategy.Subscriptions) < 1 {
+	def *integrationstypes.AWSServiceDefinition,
+) ([]*integrationstypes.SignalConnectionStatus, error) {
+	if def.Strategy == nil ||
+		len(def.Strategy.AWSMetrics.StreamFilters) < 1 ||
+		len(def.DataCollected.Metrics) < 1 {
 		return nil, nil
 	}
 
-	logGroupNamePrefix := strategy.Subscriptions[0].LogGroupNamePrefix
-	if len(logGroupNamePrefix) < 1 {
-		return nil, nil
-	}
+	statusResp := make([]*integrationstypes.SignalConnectionStatus, 0)
 
-	logsConnTestFilter := &v3.FilterSet{
-		Operator: "AND",
-		Items: []v3.FilterItem{
-			{
-				Key: v3.AttributeKey{
-					Key:      "cloud.account.id",
-					DataType: v3.AttributeKeyDataTypeString,
-					Type:     v3.AttributeKeyTypeResource,
+	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,
+					},
 				},
-				Operator: "=",
-				Value:    cloudAccountId,
-			},
-			{
-				Key: v3.AttributeKey{
-					Key:      "aws.cloudwatch.log_group_name",
-					DataType: v3.AttributeKeyDataTypeString,
-					Type:     v3.AttributeKeyTypeResource,
-				},
-				Operator: "like",
-				Value:    logGroupNamePrefix + "%",
-			},
-		},
-	}
+			})
+		}
 
-	// TODO(Raj): Receive this as a param from UI in the future.
-	lookbackSeconds := int64(30 * 60)
-
-	qrParams := &v3.QueryRangeParamsV3{
-		Start: time.Now().UnixMilli() - (lookbackSeconds * 1000),
-		End:   time.Now().UnixMilli(),
-		CompositeQuery: &v3.CompositeQuery{
-			PanelType: v3.PanelTypeList,
-			QueryType: v3.QueryTypeBuilder,
-			BuilderQueries: map[string]*v3.BuilderQuery{
-				"A": {
-					PageSize:          1,
-					Filters:           logsConnTestFilter,
-					QueryName:         "A",
-					DataSource:        v3.DataSourceLogs,
-					Expression:        "A",
-					AggregateOperator: v3.AggregateOperatorNoOp,
-				},
+		resp, err := a.querier.QueryRange(ctx, 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,
 			},
-		},
-	}
-	queryRes, _, err := a.querier.QueryRange(
-		ctx, orgID, qrParams,
-	)
-	if err != nil {
-		return nil, model.InternalError(fmt.Errorf(
-			"could not query for integration connection status: %w", err,
-		))
-	}
-	if len(queryRes) > 0 && queryRes[0].List != nil && len(queryRes[0].List) > 0 {
-		lastLog := queryRes[0].List[0]
+		})
+		if err != nil {
+			a.logger.DebugContext(ctx,
+				"error querying for service metrics connection status",
+				"error", err,
+				"service", def.DefinitionMetadata.Id,
+			)
+			continue
+		}
 
-		return &integrationstypes.SignalConnectionStatus{
-			LastReceivedTsMillis: lastLog.Timestamp.UnixMilli(),
+		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",
-		}, nil
+		})
 	}
 
-	return nil, nil
+	return statusResp, nil
+}
+
+func (a *awsProvider) getServiceLogsConnectionStatus(
+	ctx context.Context,
+	cloudAccountID string,
+	orgID valuer.UUID,
+	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, 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,

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

@@ -3,9 +3,9 @@ package cloudintegrations
 import (
 	"log/slog"
 
+	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/implawsprovider"
 	integrationstore "github.com/SigNoz/signoz/pkg/query-service/app/cloudintegrations/store"
-	"github.com/SigNoz/signoz/pkg/query-service/interfaces"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/types/integrationstypes"
 )
@@ -13,15 +13,14 @@ import (
 func NewCloudProviderRegistry(
 	logger *slog.Logger,
 	store sqlstore.SQLStore,
-	reader interfaces.Reader,
-	querier interfaces.Querier,
+	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, reader, querier)
+	awsProviderImpl := implawsprovider.NewAWSCloudProvider(logger, accountsRepo, serviceConfigRepo, querier)
 	registry[integrationstypes.CloudProviderAWS] = awsProviderImpl
 
 	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": [
       {

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": [
       {

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": [
       {

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/eks/integration.json

@@ -7,6 +7,53 @@
     "metrics": true,
     "logs": true
   },
+  "ingestion_status_check": {
+    "metrics": [
+      {
+        "category": "apiserver",
+        "display_name": "Api Server",
+        "checks": [
+          {
+            "key": "aws_EKS_apiserver_request_total_max",
+            "attributes": []
+          }
+        ]
+      },
+      {
+        "category": "container_insights",
+        "display_name": "Container Insights",
+        "checks": [
+          {
+            "key": "aws_ContainerInsights_pod_memory_utilization_max",
+            "attributes": [
+              {
+                "name": "PodName",
+                "operator": "EXISTS",
+                "value": ""
+              }
+            ]
+          }
+        ]
+      }
+    ],
+    "logs": [
+      {
+        "category": "$default",
+        "display_name": "Default",
+        "checks": [
+          {
+            "attributes": [
+              {
+                "name": "aws.cloudwatch.log_group_name",
+                "operator": "CONTAINS",
+                "value": "/aws/eks/"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
   "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":[
       {

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": [
         {

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": [
       {

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": [
       {

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": [
       {

pkg/query-service/app/http_handler.go

@@ -217,8 +217,7 @@ func NewAPIHandler(opts APIHandlerOpts) (*APIHandler, error) {
 	cloudIntegrationsRegistry := cloudintegrations.NewCloudProviderRegistry(
 		opts.Logger,
 		opts.Signoz.SQLStore,
-		opts.Reader,
-		querier,
+		opts.Signoz.Querier,
 	)
 
 	aH := &APIHandler{

pkg/types/integrationstypes/cloudintegration.go

@@ -400,11 +400,13 @@ type GettableAWSServiceDetails struct {
 }
 
 type ServiceConnectionStatus struct {
-	Logs    *SignalConnectionStatus `json:"logs"`
-	Metrics *SignalConnectionStatus `json:"metrics"`
+	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
 }

pkg/types/integrationstypes/cloudservicedefinitions.go

@@ -30,11 +30,12 @@ var _ Definition = &AWSServiceDefinition{}
 
 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"`
+	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 {