Feat: qs api integration connection status (SigNoz#4628)

* chore: add integration attribs for connection tests and status

* chore: add connection status to integration details response

* chore: update integration lifecycle test to check for connection status too

* feat: add GetIntegrationConnectionTests to integrations manager and controller

* chore: add tests for querying integration connection status

* feat: add http API support for integration connection status

* chore: some cleanups

* chore: use PostableRule for integration alerts

* chore: some more cleanup

pkg/query-service/app/clickhouseReader/reader.go

@@ -143,6 +143,17 @@ func NewReader(
 		os.Exit(1)
 	}
 
+	return NewReaderFromClickhouseConnection(db, options, localDB, configFile, featureFlag, cluster)
+}
+
+func NewReaderFromClickhouseConnection(
+	db driver.Conn,
+	options *Options,
+	localDB *sqlx.DB,
+	configFile string,
+	featureFlag interfaces.FeatureLookup,
+	cluster string,
+) *ClickHouseReader {
 	alertManager, err := am.New("")
 	if err != nil {
 		zap.S().Errorf("msg: failed to initialize alert manager: ", "/t error:", err)

pkg/query-service/app/http_handler.go

@@ -2412,6 +2412,11 @@ func (ah *APIHandler) RegisterIntegrationRoutes(router *mux.Router, am *AuthMidd
 		"/uninstall", am.ViewAccess(ah.UninstallIntegration),
 	).Methods(http.MethodPost)
 
+	// Used for polling for status in v0
+	subRouter.HandleFunc(
+		"/{integrationId}/connection_status", am.ViewAccess(ah.GetIntegrationConnectionStatus),
+	).Methods(http.MethodGet)
+
 	subRouter.HandleFunc(
 		"/{integrationId}", am.ViewAccess(ah.GetIntegration),
 	).Methods(http.MethodGet)
@@ -2443,14 +2448,112 @@ func (ah *APIHandler) GetIntegration(
 	w http.ResponseWriter, r *http.Request,
 ) {
 	integrationId := mux.Vars(r)["integrationId"]
-	resp, apiErr := ah.IntegrationsController.GetIntegration(
+	integration, apiErr := ah.IntegrationsController.GetIntegration(
 		r.Context(), integrationId,
 	)
 	if apiErr != nil {
 		RespondError(w, apiErr, "Failed to fetch integration details")
 		return
 	}
-	ah.Respond(w, resp)
+
+	// Add connection status details.
+	connectionStatus, apiErr := ah.calculateConnectionStatus(
+		r.Context(), integration.ConnectionTests,
+	)
+	if apiErr != nil {
+		RespondError(w, apiErr, "Failed to calculate integration connection status")
+		return
+	}
+	integration.ConnectionStatus = connectionStatus
+
+	ah.Respond(w, integration)
+}
+
+func (ah *APIHandler) GetIntegrationConnectionStatus(
+	w http.ResponseWriter, r *http.Request,
+) {
+	integrationId := mux.Vars(r)["integrationId"]
+	connectionTests, apiErr := ah.IntegrationsController.GetIntegrationConnectionTests(
+		r.Context(), integrationId,
+	)
+	if apiErr != nil {
+		RespondError(w, apiErr, "Failed to fetch integration connection tests")
+		return
+	}
+
+	connectionStatus, apiErr := ah.calculateConnectionStatus(
+		r.Context(), connectionTests,
+	)
+	if apiErr != nil {
+		RespondError(w, apiErr, "Failed to calculate integration connection status")
+		return
+	}
+
+	ah.Respond(w, connectionStatus)
+}
+
+func (ah *APIHandler) calculateConnectionStatus(
+	ctx context.Context,
+	connectionTests *integrations.IntegrationConnectionTests,
+) (*integrations.IntegrationConnectionStatus, *model.ApiError) {
+	result := &integrations.IntegrationConnectionStatus{}
+
+	if connectionTests.Logs != nil {
+		qrParams := &v3.QueryRangeParamsV3{
+			// Look back up to 7 days for integration logs
+			Start: time.Now().UnixMilli() - (7 * 86400000),
+			End:   time.Now().UnixMilli(),
+			CompositeQuery: &v3.CompositeQuery{
+				PanelType: v3.PanelTypeList,
+				QueryType: v3.QueryTypeBuilder,
+				BuilderQueries: map[string]*v3.BuilderQuery{
+					"A": {
+						PageSize:          1,
+						Filters:           connectionTests.Logs,
+						QueryName:         "A",
+						DataSource:        v3.DataSourceLogs,
+						Expression:        "A",
+						AggregateOperator: v3.AggregateOperatorNoOp,
+					},
+				},
+			},
+		}
+		queryRes, err, _ := ah.querier.QueryRange(
+			ctx, qrParams, map[string]v3.AttributeKey{},
+		)
+		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]
+
+			resourceSummaryParts := []string{}
+			lastLogResourceAttribs := lastLog.Data["resources_string"]
+			if lastLogResourceAttribs != nil {
+				resourceAttribs, ok := lastLogResourceAttribs.(*map[string]string)
+				if !ok {
+					return nil, model.InternalError(fmt.Errorf(
+						"could not cast log resource attribs",
+					))
+				}
+				for k, v := range *resourceAttribs {
+					resourceSummaryParts = append(resourceSummaryParts, fmt.Sprintf(
+						"%s=%s", k, v,
+					))
+				}
+			}
+			lastLogResourceSummary := strings.Join(resourceSummaryParts, ", ")
+
+			result.Logs = &integrations.SignalConnectionStatus{
+				LastReceivedTsMillis: lastLog.Timestamp.UnixMilli(),
+				LastReceivedFrom:     lastLogResourceSummary,
+			}
+		}
+	}
+
+	return result, nil
 }
 
 func (ah *APIHandler) InstallIntegration(

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

@@ -60,6 +60,12 @@ func (c *Controller) GetIntegration(
 	return c.mgr.GetIntegration(ctx, integrationId)
 }
 
+func (c *Controller) GetIntegrationConnectionTests(
+	ctx context.Context, integrationId string,
+) (*IntegrationConnectionTests, *model.ApiError) {
+	return c.mgr.GetIntegrationConnectionTests(ctx, integrationId)
+}
+
 type InstallIntegrationRequest struct {
 	IntegrationId string                 `json:"integration_id"`
 	Config        map[string]interface{} `json:"config"`

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

@@ -11,6 +11,8 @@ import (
 	"go.signoz.io/signoz/pkg/query-service/app/dashboards"
 	"go.signoz.io/signoz/pkg/query-service/app/logparsingpipeline"
 	"go.signoz.io/signoz/pkg/query-service/model"
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+	"go.signoz.io/signoz/pkg/query-service/rules"
 )
 
 type IntegrationAuthor struct {
@@ -32,8 +34,7 @@ type IntegrationAssets struct {
 	Logs       LogsAssets             `json:"logs"`
 	Dashboards []dashboards.Dashboard `json:"dashboards"`
 
-	// TODO(Raj): Maybe use a struct for alerts
-	Alerts []map[string]interface{} `json:"alerts"`
+	Alerts []rules.PostableRule `json:"alerts"`
 }
 
 type LogsAssets struct {
@@ -62,6 +63,22 @@ type CollectedMetric struct {
 	Unit string `json:"unit"`
 }
 
+type SignalConnectionStatus struct {
+	LastReceivedTsMillis int64  `json:"last_received_ts_ms"` // epoch milliseconds
+	LastReceivedFrom     string `json:"last_received_from"`  // resource identifier
+}
+
+type IntegrationConnectionStatus struct {
+	Logs    *SignalConnectionStatus `json:"logs"`
+	Metrics *SignalConnectionStatus `json:"metrics"`
+}
+
+type IntegrationConnectionTests struct {
+	Logs *v3.FilterSet `json:"logs"`
+
+	// TODO(Raj): Add connection tests for other signals.
+}
+
 type IntegrationDetails struct {
 	IntegrationSummary
 
@@ -70,6 +87,10 @@ type IntegrationDetails struct {
 	Configuration []IntegrationConfigStep     `json:"configuration"`
 	DataCollected DataCollectedForIntegration `json:"data_collected"`
 	Assets        IntegrationAssets           `json:"assets"`
+
+	ConnectionTests *IntegrationConnectionTests `json:"connection_tests"`
+	// ConnectionStatus gets derived using `ConnectionTests`
+	ConnectionStatus *IntegrationConnectionStatus `json:"connection_status"`
 }
 
 type IntegrationsListItem struct {
@@ -183,6 +204,19 @@ func (m *Manager) GetIntegration(
 	}, nil
 }
 
+func (m *Manager) GetIntegrationConnectionTests(
+	ctx context.Context,
+	integrationId string,
+) (*IntegrationConnectionTests, *model.ApiError) {
+	integrationDetails, apiErr := m.getIntegrationDetails(
+		ctx, integrationId,
+	)
+	if apiErr != nil {
+		return nil, apiErr
+	}
+	return integrationDetails.ConnectionTests, nil
+}
+
 func (m *Manager) InstallIntegration(
 	ctx context.Context,
 	integrationId string,

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

@@ -11,6 +11,7 @@ import (
 	"go.signoz.io/signoz/pkg/query-service/app/logparsingpipeline"
 	"go.signoz.io/signoz/pkg/query-service/model"
 	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+	"go.signoz.io/signoz/pkg/query-service/rules"
 )
 
 func NewTestSqliteDB(t *testing.T) (
@@ -88,12 +89,12 @@ func (t *TestAvailableIntegrationsRepo) list(
 								Items: []v3.FilterItem{
 									{
 										Key: v3.AttributeKey{
-											Key:      "method",
+											Key:      "source",
 											DataType: v3.AttributeKeyDataTypeString,
 											Type:     v3.AttributeKeyTypeTag,
 										},
 										Operator: "=",
-										Value:    "GET",
+										Value:    "nginx",
 									},
 								},
 							},
@@ -112,7 +113,23 @@ func (t *TestAvailableIntegrationsRepo) list(
 					},
 				},
 				Dashboards: []dashboards.Dashboard{},
-				Alerts:     []map[string]interface{}{},
+				Alerts:     []rules.PostableRule{},
+			},
+			ConnectionTests: &IntegrationConnectionTests{
+				Logs: &v3.FilterSet{
+					Operator: "AND",
+					Items: []v3.FilterItem{
+						{
+							Key: v3.AttributeKey{
+								Key:      "source",
+								DataType: v3.AttributeKeyDataTypeString,
+								Type:     v3.AttributeKeyTypeTag,
+							},
+							Operator: "=",
+							Value:    "nginx",
+						},
+					},
+				},
 			},
 		}, {
 			IntegrationSummary: IntegrationSummary{
@@ -150,12 +167,12 @@ func (t *TestAvailableIntegrationsRepo) list(
 								Items: []v3.FilterItem{
 									{
 										Key: v3.AttributeKey{
-											Key:      "method",
+											Key:      "source",
 											DataType: v3.AttributeKeyDataTypeString,
 											Type:     v3.AttributeKeyTypeTag,
 										},
 										Operator: "=",
-										Value:    "GET",
+										Value:    "redis",
 									},
 								},
 							},
@@ -174,7 +191,23 @@ func (t *TestAvailableIntegrationsRepo) list(
 					},
 				},
 				Dashboards: []dashboards.Dashboard{},
-				Alerts:     []map[string]interface{}{},
+				Alerts:     []rules.PostableRule{},
+			},
+			ConnectionTests: &IntegrationConnectionTests{
+				Logs: &v3.FilterSet{
+					Operator: "AND",
+					Items: []v3.FilterItem{
+						{
+							Key: v3.AttributeKey{
+								Key:      "source",
+								DataType: v3.AttributeKeyDataTypeString,
+								Type:     v3.AttributeKeyTypeTag,
+							},
+							Operator: "=",
+							Value:    "nginx",
+						},
+					},
+				},
 			},
 		},
 	}, nil