diff --git a/pkg/query-service/app/cloudintegrations/repo.go b/pkg/query-service/app/cloudintegrations/accountsRepo.go similarity index 96% rename from pkg/query-service/app/cloudintegrations/repo.go rename to pkg/query-service/app/cloudintegrations/accountsRepo.go index e5c4e4cc94..edb72209f7 100644 --- a/pkg/query-service/app/cloudintegrations/repo.go +++ b/pkg/query-service/app/cloudintegrations/accountsRepo.go @@ -37,8 +37,8 @@ type cloudProviderAccountsRepository interface { func newCloudProviderAccountsRepository(db *sqlx.DB) ( *cloudProviderAccountsSQLRepository, error, ) { - if err := InitSqliteDBIfNeeded(db); err != nil { - return nil, fmt.Errorf("could not init sqlite DB for cloudintegrations: %w", err) + if err := initAccountsSqliteDBIfNeeded(db); err != nil { + return nil, fmt.Errorf("could not init sqlite DB for cloudintegrations accounts: %w", err) } return &cloudProviderAccountsSQLRepository{ @@ -46,7 +46,7 @@ func newCloudProviderAccountsRepository(db *sqlx.DB) ( }, nil } -func InitSqliteDBIfNeeded(db *sqlx.DB) error { +func initAccountsSqliteDBIfNeeded(db *sqlx.DB) error { if db == nil { return fmt.Errorf("db is required") } @@ -66,7 +66,7 @@ func InitSqliteDBIfNeeded(db *sqlx.DB) error { _, err := db.Exec(createTablesStatements) if err != nil { return fmt.Errorf( - "could not ensure cloud provider integrations schema in sqlite DB: %w", err, + "could not ensure cloud provider accounts schema in sqlite DB: %w", err, ) } diff --git a/pkg/query-service/app/cloudintegrations/availableServices.go b/pkg/query-service/app/cloudintegrations/availableServices.go new file mode 100644 index 0000000000..474077e7e4 --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/availableServices.go @@ -0,0 +1,217 @@ +package cloudintegrations + +import ( + "bytes" + "embed" + "encoding/json" + "fmt" + "io/fs" + "path" + "sort" + + koanfJson "github.com/knadh/koanf/parsers/json" + "go.signoz.io/signoz/pkg/query-service/app/integrations" + "go.signoz.io/signoz/pkg/query-service/model" + "golang.org/x/exp/maps" +) + +func listCloudProviderServices( + cloudProvider string, +) ([]CloudServiceDetails, *model.ApiError) { + cloudServices := availableServices[cloudProvider] + if cloudServices == nil { + return nil, model.NotFoundError(fmt.Errorf( + "unsupported cloud provider: %s", cloudProvider, + )) + } + + services := maps.Values(cloudServices) + sort.Slice(services, func(i, j int) bool { + return services[i].Id < services[j].Id + }) + + return services, nil +} + +func getCloudProviderService( + cloudProvider string, serviceId string, +) (*CloudServiceDetails, *model.ApiError) { + cloudServices := availableServices[cloudProvider] + if cloudServices == nil { + return nil, model.NotFoundError(fmt.Errorf( + "unsupported cloud provider: %s", cloudProvider, + )) + } + + svc, exists := cloudServices[serviceId] + if !exists { + return nil, model.NotFoundError(fmt.Errorf( + "%s service not found: %s", cloudProvider, serviceId, + )) + } + + return &svc, nil +} + +// End of API. Logic for reading service definition files follows + +// Service details read from ./serviceDefinitions +// { "providerName": { "service_id": {...}} } +var availableServices map[string]map[string]CloudServiceDetails + +func init() { + err := readAllServiceDefinitions() + if err != nil { + panic(fmt.Errorf( + "couldn't read cloud service definitions: %w", err, + )) + } +} + +//go:embed serviceDefinitions/* +var serviceDefinitionFiles embed.FS + +func readAllServiceDefinitions() error { + availableServices = map[string]map[string]CloudServiceDetails{} + + rootDirName := "serviceDefinitions" + + cloudProviderDirs, err := fs.ReadDir(serviceDefinitionFiles, rootDirName) + if err != nil { + return fmt.Errorf("couldn't read dirs in %s: %w", rootDirName, err) + } + + for _, d := range cloudProviderDirs { + if !d.IsDir() { + continue + } + + cloudProviderDirPath := path.Join(rootDirName, d.Name()) + cloudServices, err := readServiceDefinitionsFromDir(cloudProviderDirPath) + if err != nil { + return fmt.Errorf("couldn't read %s service definitions", d.Name()) + } + + if len(cloudServices) < 1 { + return fmt.Errorf("no %s services could be read", d.Name()) + } + + availableServices[d.Name()] = cloudServices + } + + return nil +} + +func readServiceDefinitionsFromDir(cloudProviderDirPath string) ( + map[string]CloudServiceDetails, error, +) { + svcDefDirs, err := fs.ReadDir(serviceDefinitionFiles, cloudProviderDirPath) + if err != nil { + return nil, fmt.Errorf("couldn't list integrations dirs: %w", err) + } + + svcDefs := map[string]CloudServiceDetails{} + + for _, d := range svcDefDirs { + if !d.IsDir() { + continue + } + + svcDirPath := path.Join(cloudProviderDirPath, d.Name()) + s, err := readServiceDefinition(svcDirPath) + if err != nil { + return nil, fmt.Errorf("couldn't read svc definition for %s: %w", d.Name(), err) + } + + _, exists := svcDefs[s.Id] + if exists { + return nil, fmt.Errorf( + "duplicate service definition for id %s at %s", s.Id, d.Name(), + ) + } + svcDefs[s.Id] = *s + } + + return svcDefs, nil +} + +func readServiceDefinition(dirpath string) (*CloudServiceDetails, error) { + integrationJsonPath := path.Join(dirpath, "integration.json") + + serializedSpec, err := serviceDefinitionFiles.ReadFile(integrationJsonPath) + if err != nil { + return nil, fmt.Errorf( + "couldn't find integration.json in %s: %w", + dirpath, err, + ) + } + + integrationSpec, err := koanfJson.Parser().Unmarshal(serializedSpec) + if err != nil { + return nil, fmt.Errorf( + "couldn't parse integration.json from %s: %w", + integrationJsonPath, err, + ) + } + + hydrated, err := integrations.HydrateFileUris( + integrationSpec, serviceDefinitionFiles, dirpath, + ) + if err != nil { + return nil, fmt.Errorf( + "couldn't hydrate files referenced in service definition %s: %w", + integrationJsonPath, err, + ) + } + + hydratedSpec := hydrated.(map[string]interface{}) + hydratedSpecJson, err := koanfJson.Parser().Marshal(hydratedSpec) + if err != nil { + return nil, fmt.Errorf( + "couldn't serialize hydrated integration spec back to JSON %s: %w", + integrationJsonPath, err, + ) + } + + var serviceDef CloudServiceDetails + decoder := json.NewDecoder(bytes.NewReader(hydratedSpecJson)) + decoder.DisallowUnknownFields() + err = decoder.Decode(&serviceDef) + if err != nil { + return nil, fmt.Errorf( + "couldn't parse hydrated JSON spec read from %s: %w", + integrationJsonPath, err, + ) + } + + err = validateServiceDefinition(serviceDef) + if err != nil { + return nil, fmt.Errorf("invalid service definition %s: %w", serviceDef.Id, err) + } + + return &serviceDef, nil + +} + +func validateServiceDefinition(s CloudServiceDetails) error { + // Validate dashboard data + seenDashboardIds := map[string]interface{}{} + for _, dd := range s.Assets.Dashboards { + did, exists := dd["id"] + if !exists { + return fmt.Errorf("id is required. not specified in dashboard titled %v", dd["title"]) + } + dashboardId, ok := did.(string) + if !ok { + return fmt.Errorf("id must be string in dashboard titled %v", dd["title"]) + } + if _, seen := seenDashboardIds[dashboardId]; seen { + return fmt.Errorf("multiple dashboards found with id %s", dashboardId) + } + seenDashboardIds[dashboardId] = nil + } + + // potentially more to follow + + return nil +} diff --git a/pkg/query-service/app/cloudintegrations/availableServices_test.go b/pkg/query-service/app/cloudintegrations/availableServices_test.go new file mode 100644 index 0000000000..cdc222463a --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/availableServices_test.go @@ -0,0 +1,34 @@ +package cloudintegrations + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.signoz.io/signoz/pkg/query-service/model" +) + +func TestAvailableServices(t *testing.T) { + require := require.New(t) + + // should be able to list available services. + _, apiErr := listCloudProviderServices("bad-cloud-provider") + require.NotNil(apiErr) + require.Equal(model.ErrorNotFound, apiErr.Type()) + + awsSvcs, apiErr := listCloudProviderServices("aws") + require.Nil(apiErr) + require.Greater(len(awsSvcs), 0) + + // should be able to get details of a service + _, apiErr = getCloudProviderService( + "aws", "bad-service-id", + ) + require.NotNil(apiErr) + require.Equal(model.ErrorNotFound, apiErr.Type()) + + svc, apiErr := getCloudProviderService( + "aws", awsSvcs[0].Id, + ) + require.Nil(apiErr) + require.Equal(*svc, awsSvcs[0]) +} diff --git a/pkg/query-service/app/cloudintegrations/controller.go b/pkg/query-service/app/cloudintegrations/controller.go index 9cfe9fb3e1..23b8c9ac74 100644 --- a/pkg/query-service/app/cloudintegrations/controller.go +++ b/pkg/query-service/app/cloudintegrations/controller.go @@ -22,19 +22,26 @@ func validateCloudProviderName(name string) *model.ApiError { } type Controller struct { - repo cloudProviderAccountsRepository + accountsRepo cloudProviderAccountsRepository + serviceConfigRepo serviceConfigRepository } func NewController(db *sqlx.DB) ( *Controller, error, ) { - repo, err := newCloudProviderAccountsRepository(db) + accountsRepo, err := newCloudProviderAccountsRepository(db) if err != nil { return nil, fmt.Errorf("couldn't create cloud provider accounts repo: %w", err) } + serviceConfigRepo, err := newServiceConfigRepository(db) + if err != nil { + return nil, fmt.Errorf("couldn't create cloud provider service config repo: %w", err) + } + return &Controller{ - repo: repo, + accountsRepo: accountsRepo, + serviceConfigRepo: serviceConfigRepo, }, nil } @@ -58,7 +65,7 @@ func (c *Controller) ListConnectedAccounts( return nil, apiErr } - accountRecords, apiErr := c.repo.listConnected(ctx, cloudProvider) + accountRecords, apiErr := c.accountsRepo.listConnected(ctx, cloudProvider) if apiErr != nil { return nil, model.WrapApiError(apiErr, "couldn't list cloud accounts") } @@ -100,7 +107,7 @@ func (c *Controller) GenerateConnectionUrl( return nil, model.BadRequest(fmt.Errorf("unsupported cloud provider: %s", cloudProvider)) } - account, apiErr := c.repo.upsert( + account, apiErr := c.accountsRepo.upsert( ctx, cloudProvider, req.AccountId, &req.AccountConfig, nil, nil, nil, ) if apiErr != nil { @@ -133,7 +140,7 @@ func (c *Controller) GetAccountStatus( return nil, apiErr } - account, apiErr := c.repo.get(ctx, cloudProvider, accountId) + account, apiErr := c.accountsRepo.get(ctx, cloudProvider, accountId) if apiErr != nil { return nil, apiErr } @@ -164,7 +171,7 @@ func (c *Controller) CheckInAsAgent( return nil, apiErr } - existingAccount, apiErr := c.repo.get(ctx, cloudProvider, req.AccountId) + existingAccount, apiErr := c.accountsRepo.get(ctx, cloudProvider, req.AccountId) if existingAccount != nil && existingAccount.CloudAccountId != nil && *existingAccount.CloudAccountId != req.CloudAccountId { return nil, model.BadRequest(fmt.Errorf( "can't check in with new %s account id %s for account %s with existing %s id %s", @@ -172,7 +179,7 @@ func (c *Controller) CheckInAsAgent( )) } - existingAccount, apiErr = c.repo.getConnectedCloudAccount(ctx, cloudProvider, req.CloudAccountId) + existingAccount, apiErr = c.accountsRepo.getConnectedCloudAccount(ctx, cloudProvider, req.CloudAccountId) if existingAccount != nil && existingAccount.Id != req.AccountId { return nil, model.BadRequest(fmt.Errorf( "can't check in to %s account %s with id %s. already connected with id %s", @@ -185,7 +192,7 @@ func (c *Controller) CheckInAsAgent( Data: req.Data, } - account, apiErr := c.repo.upsert( + account, apiErr := c.accountsRepo.upsert( ctx, cloudProvider, &req.AccountId, nil, &req.CloudAccountId, &agentReport, nil, ) if apiErr != nil { @@ -211,7 +218,7 @@ func (c *Controller) UpdateAccountConfig( return nil, apiErr } - accountRecord, apiErr := c.repo.upsert( + accountRecord, apiErr := c.accountsRepo.upsert( ctx, cloudProvider, &accountId, &req.Config, nil, nil, nil, ) if apiErr != nil { @@ -230,13 +237,13 @@ func (c *Controller) DisconnectAccount( return nil, apiErr } - account, apiErr := c.repo.get(ctx, cloudProvider, accountId) + account, apiErr := c.accountsRepo.get(ctx, cloudProvider, accountId) if apiErr != nil { return nil, model.WrapApiError(apiErr, "couldn't disconnect account") } tsNow := time.Now() - account, apiErr = c.repo.upsert( + account, apiErr = c.accountsRepo.upsert( ctx, cloudProvider, &accountId, nil, nil, nil, &tsNow, ) if apiErr != nil { @@ -245,3 +252,127 @@ func (c *Controller) DisconnectAccount( return account, nil } + +type ListServicesResponse struct { + Services []CloudServiceSummary `json:"services"` +} + +func (c *Controller) ListServices( + ctx context.Context, + cloudProvider string, + cloudAccountId *string, +) (*ListServicesResponse, *model.ApiError) { + + if apiErr := validateCloudProviderName(cloudProvider); apiErr != nil { + return nil, apiErr + } + + services, apiErr := listCloudProviderServices(cloudProvider) + if apiErr != nil { + return nil, model.WrapApiError(apiErr, "couldn't list cloud services") + } + + svcConfigs := map[string]*CloudServiceConfig{} + if cloudAccountId != nil { + svcConfigs, apiErr = c.serviceConfigRepo.getAllForAccount( + ctx, cloudProvider, *cloudAccountId, + ) + if apiErr != nil { + return nil, model.WrapApiError( + apiErr, "couldn't get service configs for cloud account", + ) + } + } + + summaries := []CloudServiceSummary{} + for _, s := range services { + summary := s.CloudServiceSummary + summary.Config = svcConfigs[summary.Id] + + summaries = append(summaries, summary) + } + + return &ListServicesResponse{ + Services: summaries, + }, nil +} + +func (c *Controller) GetServiceDetails( + ctx context.Context, + cloudProvider string, + serviceId string, + cloudAccountId *string, +) (*CloudServiceDetails, *model.ApiError) { + + if apiErr := validateCloudProviderName(cloudProvider); apiErr != nil { + return nil, apiErr + } + + service, apiErr := getCloudProviderService(cloudProvider, serviceId) + if apiErr != nil { + return nil, apiErr + } + + if cloudAccountId != nil { + config, apiErr := c.serviceConfigRepo.get( + ctx, cloudProvider, *cloudAccountId, serviceId, + ) + if apiErr != nil && apiErr.Type() != model.ErrorNotFound { + return nil, model.WrapApiError(apiErr, "couldn't fetch service config") + } + + if config != nil { + service.Config = config + } + } + + return service, nil +} + +type UpdateServiceConfigRequest struct { + CloudAccountId string `json:"cloud_account_id"` + Config CloudServiceConfig `json:"config"` +} + +type UpdateServiceConfigResponse struct { + Id string `json:"id"` + Config CloudServiceConfig `json:"config"` +} + +func (c *Controller) UpdateServiceConfig( + ctx context.Context, + cloudProvider string, + serviceId string, + req UpdateServiceConfigRequest, +) (*UpdateServiceConfigResponse, *model.ApiError) { + + if apiErr := validateCloudProviderName(cloudProvider); apiErr != nil { + return nil, apiErr + } + + // can only update config for a connected cloud account id + _, apiErr := c.accountsRepo.getConnectedCloudAccount( + ctx, cloudProvider, req.CloudAccountId, + ) + if apiErr != nil { + return nil, model.WrapApiError(apiErr, "couldn't find connected cloud account") + } + + // can only update config for a valid service. + _, apiErr = getCloudProviderService(cloudProvider, serviceId) + if apiErr != nil { + return nil, model.WrapApiError(apiErr, "unsupported service") + } + + updatedConfig, apiErr := c.serviceConfigRepo.upsert( + ctx, cloudProvider, req.CloudAccountId, serviceId, req.Config, + ) + if apiErr != nil { + return nil, model.WrapApiError(apiErr, "couldn't update service config") + } + + return &UpdateServiceConfigResponse{ + Id: serviceId, + Config: *updatedConfig, + }, nil +} diff --git a/pkg/query-service/app/cloudintegrations/controller_test.go b/pkg/query-service/app/cloudintegrations/controller_test.go index 2d85cc1b95..7b361615d0 100644 --- a/pkg/query-service/app/cloudintegrations/controller_test.go +++ b/pkg/query-service/app/cloudintegrations/controller_test.go @@ -30,7 +30,7 @@ func TestRegenerateConnectionUrlWithUpdatedConfig(t *testing.T) { require.NotEmpty(resp1.AccountId) testAccountId := resp1.AccountId - account, apiErr := controller.repo.get( + account, apiErr := controller.accountsRepo.get( context.TODO(), "aws", testAccountId, ) require.Nil(apiErr) @@ -47,7 +47,7 @@ func TestRegenerateConnectionUrlWithUpdatedConfig(t *testing.T) { require.Nil(apiErr) require.Equal(testAccountId, resp2.AccountId) - account, apiErr = controller.repo.get( + account, apiErr = controller.accountsRepo.get( context.TODO(), "aws", testAccountId, ) require.Nil(apiErr) @@ -89,7 +89,7 @@ func TestAgentCheckIns(t *testing.T) { // if another connected AccountRecord exists for same cloud account // i.e. there can't be 2 connected account records for the same cloud account id // at any point in time. - existingConnected, apiErr := controller.repo.getConnectedCloudAccount( + existingConnected, apiErr := controller.accountsRepo.getConnectedCloudAccount( context.TODO(), "aws", testCloudAccountId1, ) require.Nil(apiErr) @@ -112,7 +112,7 @@ func TestAgentCheckIns(t *testing.T) { context.TODO(), "aws", testAccountId1, ) - existingConnected, apiErr = controller.repo.getConnectedCloudAccount( + existingConnected, apiErr = controller.accountsRepo.getConnectedCloudAccount( context.TODO(), "aws", testCloudAccountId1, ) require.Nil(existingConnected) @@ -151,3 +151,120 @@ func TestCantDisconnectNonExistentAccount(t *testing.T) { require.Equal(model.ErrorNotFound, apiErr.Type()) require.Nil(account) } + +func TestConfigureService(t *testing.T) { + require := require.New(t) + testDB, _ := utils.NewTestSqliteDB(t) + controller, err := NewController(testDB) + require.NoError(err) + + testCloudAccountId := "546311234" + + // should start out without any service config + svcListResp, apiErr := controller.ListServices( + context.TODO(), "aws", &testCloudAccountId, + ) + require.Nil(apiErr) + + testSvcId := svcListResp.Services[0].Id + require.Nil(svcListResp.Services[0].Config) + + svcDetails, apiErr := controller.GetServiceDetails( + context.TODO(), "aws", testSvcId, &testCloudAccountId, + ) + require.Nil(apiErr) + require.Equal(testSvcId, svcDetails.Id) + require.Nil(svcDetails.Config) + + // should be able to configure a service for a connected account + testConnectedAccount := makeTestConnectedAccount(t, controller, testCloudAccountId) + require.Nil(testConnectedAccount.RemovedAt) + require.NotNil(testConnectedAccount.CloudAccountId) + require.Equal(testCloudAccountId, *testConnectedAccount.CloudAccountId) + + testSvcConfig := CloudServiceConfig{ + Metrics: &CloudServiceMetricsConfig{ + Enabled: true, + }, + } + updateSvcConfigResp, apiErr := controller.UpdateServiceConfig( + context.TODO(), "aws", testSvcId, UpdateServiceConfigRequest{ + CloudAccountId: testCloudAccountId, + Config: testSvcConfig, + }, + ) + require.Nil(apiErr) + require.Equal(testSvcId, updateSvcConfigResp.Id) + require.Equal(testSvcConfig, updateSvcConfigResp.Config) + + svcDetails, apiErr = controller.GetServiceDetails( + context.TODO(), "aws", testSvcId, &testCloudAccountId, + ) + require.Nil(apiErr) + require.Equal(testSvcId, svcDetails.Id) + require.Equal(testSvcConfig, *svcDetails.Config) + + svcListResp, apiErr = controller.ListServices( + context.TODO(), "aws", &testCloudAccountId, + ) + require.Nil(apiErr) + for _, svc := range svcListResp.Services { + if svc.Id == testSvcId { + require.Equal(testSvcConfig, *svc.Config) + } + } + + // should not be able to configure service after cloud account has been disconnected + _, apiErr = controller.DisconnectAccount( + context.TODO(), "aws", testConnectedAccount.Id, + ) + require.Nil(apiErr) + + _, apiErr = controller.UpdateServiceConfig( + context.TODO(), "aws", testSvcId, + UpdateServiceConfigRequest{ + CloudAccountId: testCloudAccountId, + Config: testSvcConfig, + }, + ) + require.NotNil(apiErr) + + // should not be able to configure a service for a cloud account id that is not connected yet + _, apiErr = controller.UpdateServiceConfig( + context.TODO(), "aws", testSvcId, + UpdateServiceConfigRequest{ + CloudAccountId: "9999999999", + Config: testSvcConfig, + }, + ) + require.NotNil(apiErr) + + // should not be able to set config for an unsupported service + _, apiErr = controller.UpdateServiceConfig( + context.TODO(), "aws", "bad-service", UpdateServiceConfigRequest{ + CloudAccountId: testCloudAccountId, + Config: testSvcConfig, + }, + ) + require.NotNil(apiErr) + +} + +func makeTestConnectedAccount(t *testing.T, controller *Controller, cloudAccountId string) *AccountRecord { + require := require.New(t) + + // a check in from SigNoz agent creates or updates a connected account. + testAccountId := uuid.NewString() + resp, apiErr := controller.CheckInAsAgent( + context.TODO(), "aws", AgentCheckInRequest{ + AccountId: testAccountId, + CloudAccountId: cloudAccountId, + }, + ) + require.Nil(apiErr) + require.Equal(testAccountId, resp.Account.Id) + require.Equal(cloudAccountId, *resp.Account.CloudAccountId) + + return &resp.Account + +} diff --git a/pkg/query-service/app/cloudintegrations/model.go b/pkg/query-service/app/cloudintegrations/model.go index 53ad4c853e..5f35336b9f 100644 --- a/pkg/query-service/app/cloudintegrations/model.go +++ b/pkg/query-service/app/cloudintegrations/model.go @@ -5,6 +5,8 @@ import ( "encoding/json" "fmt" "time" + + "go.signoz.io/signoz/pkg/query-service/app/dashboards" ) // Represents a cloud provider account for cloud integrations @@ -115,3 +117,102 @@ func (a *AccountRecord) account() Account { return ca } + +type CloudServiceSummary struct { + Id string `json:"id"` + Title string `json:"title"` + Icon string `json:"icon"` + + // Present only if the service has been configured in the + // context of a cloud provider account. + Config *CloudServiceConfig `json:"config,omitempty"` +} + +type CloudServiceDetails struct { + CloudServiceSummary + + Overview string `json:"overview"` // markdown + + Assets CloudServiceAssets `json:"assets"` + + SupportedSignals SupportedSignals `json:"supported_signals"` + + DataCollected DataCollectedForService `json:"data_collected"` + + ConnectionStatus *CloudServiceConnectionStatus `json:"status,omitempty"` +} + +type CloudServiceConfig struct { + Logs *CloudServiceLogsConfig `json:"logs,omitempty"` + Metrics *CloudServiceMetricsConfig `json:"metrics,omitempty"` +} + +// For serializing from db +func (c *CloudServiceConfig) Scan(src any) error { + data, ok := src.([]byte) + if !ok { + return fmt.Errorf("tried to scan from %T instead of bytes", src) + } + + return json.Unmarshal(data, &c) +} + +// For serializing to db +func (c *CloudServiceConfig) Value() (driver.Value, error) { + if c == nil { + return nil, nil + } + + serialized, err := json.Marshal(c) + if err != nil { + return nil, fmt.Errorf( + "couldn't serialize cloud service config to JSON: %w", err, + ) + } + return serialized, nil +} + +type CloudServiceLogsConfig struct { + Enabled bool `json:"enabled"` +} + +type CloudServiceMetricsConfig struct { + Enabled bool `json:"enabled"` +} + +type CloudServiceAssets struct { + Dashboards []dashboards.Data `json:"dashboards"` +} + +type SupportedSignals struct { + Logs bool `json:"logs"` + Metrics bool `json:"metrics"` +} + +type DataCollectedForService 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 CloudServiceConnectionStatus 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 +} diff --git a/pkg/query-service/app/cloudintegrations/serviceConfigRepo.go b/pkg/query-service/app/cloudintegrations/serviceConfigRepo.go new file mode 100644 index 0000000000..17994e7565 --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceConfigRepo.go @@ -0,0 +1,198 @@ +package cloudintegrations + +import ( + "context" + "database/sql" + "fmt" + + "github.com/jmoiron/sqlx" + "go.signoz.io/signoz/pkg/query-service/model" +) + +type serviceConfigRepository interface { + get( + ctx context.Context, + cloudProvider string, + cloudAccountId string, + serviceId string, + ) (*CloudServiceConfig, *model.ApiError) + + upsert( + ctx context.Context, + cloudProvider string, + cloudAccountId string, + serviceId string, + config CloudServiceConfig, + ) (*CloudServiceConfig, *model.ApiError) + + getAllForAccount( + ctx context.Context, + cloudProvider string, + cloudAccountId string, + ) ( + configsBySvcId map[string]*CloudServiceConfig, + apiErr *model.ApiError, + ) +} + +func newServiceConfigRepository(db *sqlx.DB) ( + *serviceConfigSQLRepository, error, +) { + + if err := initServiceConfigSqliteDBIfNeeded(db); err != nil { + return nil, fmt.Errorf( + "could not init sqlite DB for cloudintegrations service configs: %w", err, + ) + } + + return &serviceConfigSQLRepository{ + db: db, + }, nil +} + +func initServiceConfigSqliteDBIfNeeded(db *sqlx.DB) error { + + if db == nil { + return fmt.Errorf("db is required") + } + + createTableStatement := ` + CREATE TABLE IF NOT EXISTS cloud_integrations_service_configs( + cloud_provider TEXT NOT NULL, + cloud_account_id TEXT NOT NULL, + service_id TEXT NOT NULL, + config_json TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, + UNIQUE(cloud_provider, cloud_account_id, service_id) + ) + ` + _, err := db.Exec(createTableStatement) + if err != nil { + return fmt.Errorf( + "could not ensure cloud provider service configs schema in sqlite DB: %w", err, + ) + } + + return nil +} + +type serviceConfigSQLRepository struct { + db *sqlx.DB +} + +func (r *serviceConfigSQLRepository) get( + ctx context.Context, + cloudProvider string, + cloudAccountId string, + serviceId string, +) (*CloudServiceConfig, *model.ApiError) { + + var result CloudServiceConfig + + err := r.db.GetContext( + ctx, &result, ` + select + config_json + from cloud_integrations_service_configs + where + cloud_provider=$1 + and cloud_account_id=$2 + and service_id=$3 + `, + cloudProvider, cloudAccountId, serviceId, + ) + + if err == sql.ErrNoRows { + return nil, model.NotFoundError(fmt.Errorf( + "couldn't find %s %s config for %s", + cloudProvider, serviceId, cloudAccountId, + )) + + } else if err != nil { + return nil, model.InternalError(fmt.Errorf( + "couldn't query cloud service config: %w", err, + )) + } + + return &result, nil + +} + +func (r *serviceConfigSQLRepository) upsert( + ctx context.Context, + cloudProvider string, + cloudAccountId string, + serviceId string, + config CloudServiceConfig, +) (*CloudServiceConfig, *model.ApiError) { + + query := ` + INSERT INTO cloud_integrations_service_configs ( + cloud_provider, + cloud_account_id, + service_id, + config_json + ) values ($1, $2, $3, $4) + on conflict(cloud_provider, cloud_account_id, service_id) + do update set config_json=excluded.config_json + ` + _, dbErr := r.db.ExecContext( + ctx, query, + cloudProvider, cloudAccountId, serviceId, &config, + ) + if dbErr != nil { + return nil, model.InternalError(fmt.Errorf( + "could not upsert cloud service config: %w", dbErr, + )) + } + + upsertedConfig, apiErr := r.get(ctx, cloudProvider, cloudAccountId, serviceId) + if apiErr != nil { + return nil, model.InternalError(fmt.Errorf( + "couldn't fetch upserted service config: %w", apiErr.ToError(), + )) + } + + return upsertedConfig, nil + +} + +func (r *serviceConfigSQLRepository) getAllForAccount( + ctx context.Context, + cloudProvider string, + cloudAccountId string, +) (map[string]*CloudServiceConfig, *model.ApiError) { + + type ScannedServiceConfigRecord struct { + ServiceId string `db:"service_id"` + Config CloudServiceConfig `db:"config_json"` + } + + records := []ScannedServiceConfigRecord{} + + err := r.db.SelectContext( + ctx, &records, ` + select + service_id, + config_json + from cloud_integrations_service_configs + where + cloud_provider=$1 + and cloud_account_id=$2 + `, + cloudProvider, cloudAccountId, + ) + if err != nil { + return nil, model.InternalError(fmt.Errorf( + "could not query service configs from db: %w", err, + )) + } + + result := map[string]*CloudServiceConfig{} + + for _, r := range records { + result[r.ServiceId] = &r.Config + } + + return result, nil +} diff --git a/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/icon.svg b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/icon.svg new file mode 100644 index 0000000000..6c85e7c8d2 --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/icon.svg @@ -0,0 +1,11 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/integration.json b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/integration.json new file mode 100644 index 0000000000..55f85ca5bd --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/integration.json @@ -0,0 +1,30 @@ +{ + "id": "ec2", + "title": "EC2", + "icon": "file://icon.svg", + "overview": "file://overview.md", + "assets": { + "dashboards": [] + }, + "supported_signals": { + "metrics": true, + "logs": false + }, + "data_collected": { + "metrics": [ + { + "name": "ec2_cpuutilization_average", + "type": "Gauge", + "unit": "number", + "description": "CloudWatch metric CPUUtilization" + }, + { + "name": "ec2_cpuutilization_maximum", + "type": "Gauge", + "unit": "number", + "description": "CloudWatch metric CPUUtilization" + } + ], + "logs": [] + } +} \ No newline at end of file diff --git a/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/overview.md b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/overview.md new file mode 100644 index 0000000000..3a1642c016 --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/ec2/overview.md @@ -0,0 +1,3 @@ +### Monitor EC2 with SigNoz + +Collect key EC2 metrics and view them with an out of the box dashboard. diff --git a/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/icon.svg b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/icon.svg new file mode 100644 index 0000000000..0eab74ea79 --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/icon.svg @@ -0,0 +1,21 @@ + + + Icon-Architecture/64/Arch_Amazon-RDS_64 + Created with Sketch. + + + + + + + + + + + + + \ No newline at end of file diff --git a/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/integration.json b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/integration.json new file mode 100644 index 0000000000..969cbf56b4 --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/integration.json @@ -0,0 +1,30 @@ +{ + "id": "rds-postgres", + "title": "RDS Postgres", + "icon": "file://icon.svg", + "overview": "file://overview.md", + "assets": { + "dashboards": [] + }, + "supported_signals": { + "metrics": true, + "logs": true + }, + "data_collected": { + "metrics": [ + { + "name": "rds_postgres_cpuutilization_average", + "type": "Gauge", + "unit": "number", + "description": "CloudWatch metric CPUUtilization" + }, + { + "name": "rds_postgres_cpuutilization_maximum", + "type": "Gauge", + "unit": "number", + "description": "CloudWatch metric CPUUtilization" + } + ], + "logs": [] + } +} \ No newline at end of file diff --git a/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/overview.md b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/overview.md new file mode 100644 index 0000000000..193353be1f --- /dev/null +++ b/pkg/query-service/app/cloudintegrations/serviceDefinitions/aws/rdsPostgres/overview.md @@ -0,0 +1,3 @@ +### Monitor RDS Postgres with SigNoz + +Collect key RDS Postgres metrics and view them with an out of the box dashboard. diff --git a/pkg/query-service/app/http_handler.go b/pkg/query-service/app/http_handler.go index 5da9200c0e..fff4fded57 100644 --- a/pkg/query-service/app/http_handler.go +++ b/pkg/query-service/app/http_handler.go @@ -3902,6 +3902,18 @@ func (aH *APIHandler) RegisterCloudIntegrationsRoutes(router *mux.Router, am *Au "/{cloudProvider}/agent-check-in", am.EditAccess(aH.CloudIntegrationsAgentCheckIn), ).Methods(http.MethodPost) + subRouter.HandleFunc( + "/{cloudProvider}/services", am.ViewAccess(aH.CloudIntegrationsListServices), + ).Methods(http.MethodGet) + + subRouter.HandleFunc( + "/{cloudProvider}/services/{serviceId}", am.ViewAccess(aH.CloudIntegrationsGetServiceDetails), + ).Methods(http.MethodGet) + + subRouter.HandleFunc( + "/{cloudProvider}/services/{serviceId}/config", am.EditAccess(aH.CloudIntegrationsUpdateServiceConfig), + ).Methods(http.MethodPost) + } func (aH *APIHandler) CloudIntegrationsListConnectedAccounts( @@ -4025,6 +4037,77 @@ func (aH *APIHandler) CloudIntegrationsDisconnectAccount( aH.Respond(w, result) } +func (aH *APIHandler) CloudIntegrationsListServices( + w http.ResponseWriter, r *http.Request, +) { + cloudProvider := mux.Vars(r)["cloudProvider"] + + var cloudAccountId *string + + cloudAccountIdQP := r.URL.Query().Get("cloud_account_id") + if len(cloudAccountIdQP) > 0 { + cloudAccountId = &cloudAccountIdQP + } + + resp, apiErr := aH.CloudIntegrationsController.ListServices( + r.Context(), cloudProvider, cloudAccountId, + ) + + if apiErr != nil { + RespondError(w, apiErr, nil) + return + } + aH.Respond(w, resp) +} + +func (aH *APIHandler) CloudIntegrationsGetServiceDetails( + w http.ResponseWriter, r *http.Request, +) { + cloudProvider := mux.Vars(r)["cloudProvider"] + serviceId := mux.Vars(r)["serviceId"] + + var cloudAccountId *string + + cloudAccountIdQP := r.URL.Query().Get("cloud_account_id") + if len(cloudAccountIdQP) > 0 { + cloudAccountId = &cloudAccountIdQP + } + + resp, apiErr := aH.CloudIntegrationsController.GetServiceDetails( + r.Context(), cloudProvider, serviceId, cloudAccountId, + ) + + if apiErr != nil { + RespondError(w, apiErr, nil) + return + } + aH.Respond(w, resp) +} + +func (aH *APIHandler) CloudIntegrationsUpdateServiceConfig( + w http.ResponseWriter, r *http.Request, +) { + cloudProvider := mux.Vars(r)["cloudProvider"] + serviceId := mux.Vars(r)["serviceId"] + + req := cloudintegrations.UpdateServiceConfigRequest{} + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + RespondError(w, model.BadRequest(err), nil) + return + } + + result, apiErr := aH.CloudIntegrationsController.UpdateServiceConfig( + r.Context(), cloudProvider, serviceId, req, + ) + + if apiErr != nil { + RespondError(w, apiErr, nil) + return + } + + aH.Respond(w, result) +} + // logs func (aH *APIHandler) RegisterLogsRoutes(router *mux.Router, am *AuthMiddleware) { subRouter := router.PathPrefix("/api/v1/logs").Subrouter() diff --git a/pkg/query-service/app/integrations/builtin.go b/pkg/query-service/app/integrations/builtin.go index d8099633ab..77e31df001 100644 --- a/pkg/query-service/app/integrations/builtin.go +++ b/pkg/query-service/app/integrations/builtin.go @@ -105,7 +105,7 @@ func readBuiltInIntegration(dirpath string) ( ) } - hydrated, err := hydrateFileUris(integrationSpec, dirpath) + hydrated, err := HydrateFileUris(integrationSpec, integrationFiles, dirpath) if err != nil { return nil, fmt.Errorf( "couldn't hydrate files referenced in integration %s: %w", integrationJsonPath, err, @@ -172,11 +172,11 @@ func validateIntegration(i IntegrationDetails) error { return nil } -func hydrateFileUris(spec interface{}, basedir string) (interface{}, error) { +func HydrateFileUris(spec interface{}, fs embed.FS, basedir string) (interface{}, error) { if specMap, ok := spec.(map[string]interface{}); ok { result := map[string]interface{}{} for k, v := range specMap { - hydrated, err := hydrateFileUris(v, basedir) + hydrated, err := HydrateFileUris(v, fs, basedir) if err != nil { return nil, err } @@ -187,7 +187,7 @@ func hydrateFileUris(spec interface{}, basedir string) (interface{}, error) { } else if specSlice, ok := spec.([]interface{}); ok { result := []interface{}{} for _, v := range specSlice { - hydrated, err := hydrateFileUris(v, basedir) + hydrated, err := HydrateFileUris(v, fs, basedir) if err != nil { return nil, err } @@ -196,14 +196,14 @@ func hydrateFileUris(spec interface{}, basedir string) (interface{}, error) { return result, nil } else if maybeFileUri, ok := spec.(string); ok { - return readFileIfUri(maybeFileUri, basedir) + return readFileIfUri(fs, maybeFileUri, basedir) } return spec, nil } -func readFileIfUri(maybeFileUri string, basedir string) (interface{}, error) { +func readFileIfUri(fs embed.FS, maybeFileUri string, basedir string) (interface{}, error) { fileUriPrefix := "file://" if !strings.HasPrefix(maybeFileUri, fileUriPrefix) { return maybeFileUri, nil @@ -212,7 +212,7 @@ func readFileIfUri(maybeFileUri string, basedir string) (interface{}, error) { relativePath := maybeFileUri[len(fileUriPrefix):] fullPath := path.Join(basedir, relativePath) - fileContents, err := integrationFiles.ReadFile(fullPath) + fileContents, err := fs.ReadFile(fullPath) if err != nil { return nil, fmt.Errorf("couldn't read referenced file: %w", err) } diff --git a/pkg/query-service/tests/integration/signoz_cloud_integrations_test.go b/pkg/query-service/tests/integration/signoz_cloud_integrations_test.go index 9f199b5b06..55f337742d 100644 --- a/pkg/query-service/tests/integration/signoz_cloud_integrations_test.go +++ b/pkg/query-service/tests/integration/signoz_cloud_integrations_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/google/uuid" "github.com/jmoiron/sqlx" mockhouse "github.com/srikanthccv/ClickHouse-go-mock" "github.com/stretchr/testify/require" @@ -19,7 +20,7 @@ import ( "go.signoz.io/signoz/pkg/query-service/utils" ) -func TestAWSIntegrationLifecycle(t *testing.T) { +func TestAWSIntegrationAccountLifecycle(t *testing.T) { // Test for happy path of connecting and managing AWS integration accounts t0 := time.Now() @@ -126,6 +127,70 @@ func TestAWSIntegrationLifecycle(t *testing.T) { require.LessOrEqual(tsBeforeDisconnect, *agentCheckInResp2.Account.RemovedAt) } +func TestAWSIntegrationServices(t *testing.T) { + require := require.New(t) + + testbed := NewCloudIntegrationsTestBed(t, nil) + + // should be able to list available cloud services. + svcListResp := testbed.GetServicesFromQS("aws", nil) + require.Greater(len(svcListResp.Services), 0) + for _, svc := range svcListResp.Services { + require.NotEmpty(svc.Id) + require.Nil(svc.Config) + } + + // should be able to get details of a particular service. + svcId := svcListResp.Services[0].Id + svcDetailResp := testbed.GetServiceDetailFromQS("aws", svcId, nil) + require.Equal(svcId, svcDetailResp.Id) + require.NotEmpty(svcDetailResp.Overview) + require.Nil(svcDetailResp.Config) + require.Nil(svcDetailResp.ConnectionStatus) + + // should be able to configure a service in the ctx of a connected account + + // create a connected account + testAccountId := uuid.NewString() + testAWSAccountId := "389389489489" + testbed.CheckInAsAgentWithQS( + "aws", cloudintegrations.AgentCheckInRequest{ + AccountId: testAccountId, + CloudAccountId: testAWSAccountId, + }, + ) + + testSvcConfig := cloudintegrations.CloudServiceConfig{ + Metrics: &cloudintegrations.CloudServiceMetricsConfig{ + Enabled: true, + }, + } + updateSvcConfigResp := testbed.UpdateServiceConfigWithQS("aws", svcId, cloudintegrations.UpdateServiceConfigRequest{ + CloudAccountId: testAWSAccountId, + Config: testSvcConfig, + }) + require.Equal(svcId, updateSvcConfigResp.Id) + require.Equal(testSvcConfig, updateSvcConfigResp.Config) + + // service list should include config when queried in the ctx of an account + svcListResp = testbed.GetServicesFromQS("aws", &testAWSAccountId) + require.Greater(len(svcListResp.Services), 0) + for _, svc := range svcListResp.Services { + if svc.Id == svcId { + require.NotNil(svc.Config) + require.Equal(testSvcConfig, *svc.Config) + } + } + + // service detail should include config and status info when + // queried in the ctx of an account + svcDetailResp = testbed.GetServiceDetailFromQS("aws", svcId, &testAWSAccountId) + require.Equal(svcId, svcDetailResp.Id) + require.NotNil(svcDetailResp.Config) + require.Equal(testSvcConfig, *svcDetailResp.Config) + +} + type CloudIntegrationsTestBed struct { t *testing.T testUser *model.User @@ -275,6 +340,41 @@ func (tb *CloudIntegrationsTestBed) DisconnectAccountWithQS( return &resp } +func (tb *CloudIntegrationsTestBed) GetServicesFromQS( + cloudProvider string, cloudAccountId *string, +) *cloudintegrations.ListServicesResponse { + path := fmt.Sprintf("/api/v1/cloud-integrations/%s/services", cloudProvider) + if cloudAccountId != nil { + path = fmt.Sprintf("%s?cloud_account_id=%s", path, *cloudAccountId) + } + + return RequestQSAndParseResp[cloudintegrations.ListServicesResponse]( + tb, path, nil, + ) +} + +func (tb *CloudIntegrationsTestBed) GetServiceDetailFromQS( + cloudProvider string, serviceId string, cloudAccountId *string, +) *cloudintegrations.CloudServiceDetails { + path := fmt.Sprintf("/api/v1/cloud-integrations/%s/services/%s", cloudProvider, serviceId) + if cloudAccountId != nil { + path = fmt.Sprintf("%s?cloud_account_id=%s", path, *cloudAccountId) + } + + return RequestQSAndParseResp[cloudintegrations.CloudServiceDetails]( + tb, path, nil, + ) +} +func (tb *CloudIntegrationsTestBed) UpdateServiceConfigWithQS( + cloudProvider string, serviceId string, req any, +) *cloudintegrations.UpdateServiceConfigResponse { + path := fmt.Sprintf("/api/v1/cloud-integrations/%s/services/%s/config", cloudProvider, serviceId) + + return RequestQSAndParseResp[cloudintegrations.UpdateServiceConfigResponse]( + tb, path, req, + ) +} + func (tb *CloudIntegrationsTestBed) RequestQS( path string, postData interface{}, @@ -297,3 +397,20 @@ func (tb *CloudIntegrationsTestBed) RequestQS( } return dataJson } + +func RequestQSAndParseResp[ResponseType any]( + tb *CloudIntegrationsTestBed, + path string, + postData interface{}, +) *ResponseType { + respDataJson := tb.RequestQS(path, postData) + + var resp ResponseType + + err := json.Unmarshal(respDataJson, &resp) + if err != nil { + tb.t.Fatalf("could not unmarshal apiResponse.Data json into %T", resp) + } + + return &resp +}