mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-28 14:20:42 +01:00
Compare commits
3 Commits
v0.143.0-c
...
issue-2977
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3ae7d4441f | ||
|
|
cbfe328936 | ||
|
|
9708e89d8c |
@@ -1763,12 +1763,15 @@ components:
|
||||
additionalProperties: {}
|
||||
nullable: true
|
||||
type: object
|
||||
syncState:
|
||||
$ref: '#/components/schemas/CloudintegrationtypesSyncState'
|
||||
timestampMillis:
|
||||
format: int64
|
||||
type: integer
|
||||
required:
|
||||
- timestampMillis
|
||||
- data
|
||||
- syncState
|
||||
type: object
|
||||
CloudintegrationtypesAzureAccountConfig:
|
||||
properties:
|
||||
@@ -2013,6 +2016,8 @@ components:
|
||||
format: date-time
|
||||
nullable: true
|
||||
type: string
|
||||
syncState:
|
||||
$ref: '#/components/schemas/CloudintegrationtypesSyncState'
|
||||
required:
|
||||
- account_id
|
||||
- cloud_account_id
|
||||
@@ -2022,6 +2027,7 @@ components:
|
||||
- providerAccountId
|
||||
- integrationConfig
|
||||
- removedAt
|
||||
- syncState
|
||||
type: object
|
||||
CloudintegrationtypesGettableServicesMetadata:
|
||||
properties:
|
||||
@@ -2121,6 +2127,9 @@ components:
|
||||
type: object
|
||||
providerAccountId:
|
||||
type: string
|
||||
syncedVersion:
|
||||
nullable: true
|
||||
type: integer
|
||||
required:
|
||||
- data
|
||||
type: object
|
||||
@@ -2133,6 +2142,18 @@ components:
|
||||
gcp:
|
||||
$ref: '#/components/schemas/CloudintegrationtypesGCPIntegrationConfig'
|
||||
type: object
|
||||
CloudintegrationtypesRegionState:
|
||||
enum:
|
||||
- present
|
||||
- removed
|
||||
type: string
|
||||
CloudintegrationtypesRegionSyncState:
|
||||
properties:
|
||||
state:
|
||||
$ref: '#/components/schemas/CloudintegrationtypesRegionState'
|
||||
required:
|
||||
- state
|
||||
type: object
|
||||
CloudintegrationtypesService:
|
||||
properties:
|
||||
assets:
|
||||
@@ -2274,6 +2295,23 @@ components:
|
||||
metrics:
|
||||
type: boolean
|
||||
type: object
|
||||
CloudintegrationtypesSyncState:
|
||||
nullable: true
|
||||
properties:
|
||||
inSync:
|
||||
type: boolean
|
||||
regions:
|
||||
additionalProperties:
|
||||
$ref: '#/components/schemas/CloudintegrationtypesRegionSyncState'
|
||||
type: object
|
||||
version:
|
||||
format: int64
|
||||
type: integer
|
||||
required:
|
||||
- version
|
||||
- inSync
|
||||
- regions
|
||||
type: object
|
||||
CloudintegrationtypesUpdatableAccount:
|
||||
properties:
|
||||
config:
|
||||
|
||||
@@ -188,27 +188,41 @@ func (module *module) AgentCheckIn(ctx context.Context, orgID valuer.UUID, provi
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Get account as domain object for config access (enabled regions, etc.)
|
||||
domainAccount, err := cloudintegrationtypes.NewAccountFromStorable(account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
syncState := domainAccount.NextSyncState(req.SyncedVersion)
|
||||
|
||||
// If account has been removed (disconnected), return a minimal response with empty integration config.
|
||||
// The agent uses this response to clean up resources
|
||||
if account.RemovedAt != nil {
|
||||
// Heartbeat stays frozen after removal, only the sync state is updated.
|
||||
if domainAccount.AgentReport != nil && syncState != nil {
|
||||
domainAccount.AgentReport.SyncState = syncState
|
||||
account.Update(account.AccountID, domainAccount.AgentReport)
|
||||
|
||||
err = module.store.UpdateAgentReport(ctx, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return cloudintegrationtypes.NewAgentCheckInResponse(
|
||||
req.ProviderAccountID,
|
||||
account.ID.StringValue(),
|
||||
new(cloudintegrationtypes.ProviderIntegrationConfig),
|
||||
account.RemovedAt,
|
||||
syncState,
|
||||
), nil
|
||||
}
|
||||
|
||||
// update account with cloud provider account id and agent report (heartbeat)
|
||||
account.Update(&req.ProviderAccountID, cloudintegrationtypes.NewAgentReport(req.Data))
|
||||
account.Update(&req.ProviderAccountID, cloudintegrationtypes.NewAgentReport(req.Data, syncState))
|
||||
|
||||
err = module.store.UpdateAccount(ctx, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Get account as domain object for config access (enabled regions, etc.)
|
||||
domainAccount, err := cloudintegrationtypes.NewAccountFromStorable(account)
|
||||
err = module.store.UpdateAgentReport(ctx, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -234,6 +248,7 @@ func (module *module) AgentCheckIn(ctx context.Context, orgID valuer.UUID, provi
|
||||
account.ID.StringValue(),
|
||||
integrationConfig,
|
||||
account.RemovedAt,
|
||||
syncState,
|
||||
), nil
|
||||
}
|
||||
|
||||
|
||||
@@ -3366,6 +3366,37 @@ export interface CloudintegrationtypesAWSServiceConfigDTO {
|
||||
metrics?: CloudintegrationtypesAWSServiceMetricsConfigDTO;
|
||||
}
|
||||
|
||||
export enum CloudintegrationtypesRegionStateDTO {
|
||||
present = 'present',
|
||||
removed = 'removed',
|
||||
}
|
||||
export interface CloudintegrationtypesRegionSyncStateDTO {
|
||||
state: CloudintegrationtypesRegionStateDTO;
|
||||
}
|
||||
|
||||
export type CloudintegrationtypesSyncStateDTORegions = {
|
||||
[key: string]: CloudintegrationtypesRegionSyncStateDTO;
|
||||
};
|
||||
|
||||
/**
|
||||
* @nullable
|
||||
*/
|
||||
export type CloudintegrationtypesSyncStateDTO = {
|
||||
/**
|
||||
* @type boolean
|
||||
*/
|
||||
inSync: boolean;
|
||||
/**
|
||||
* @type object
|
||||
*/
|
||||
regions: CloudintegrationtypesSyncStateDTORegions;
|
||||
/**
|
||||
* @type integer
|
||||
* @format int64
|
||||
*/
|
||||
version: number;
|
||||
} | null;
|
||||
|
||||
export type CloudintegrationtypesAgentReportDTODataAnyOf = {
|
||||
[key: string]: unknown;
|
||||
};
|
||||
@@ -3384,6 +3415,7 @@ export type CloudintegrationtypesAgentReportDTO = {
|
||||
* @type object,null
|
||||
*/
|
||||
data: CloudintegrationtypesAgentReportDTOData;
|
||||
syncState: CloudintegrationtypesSyncStateDTO | null;
|
||||
/**
|
||||
* @type integer
|
||||
* @format int64
|
||||
@@ -3812,6 +3844,7 @@ export interface CloudintegrationtypesGettableAgentCheckInDTO {
|
||||
* @format date-time
|
||||
*/
|
||||
removedAt: string | null;
|
||||
syncState: CloudintegrationtypesSyncStateDTO | null;
|
||||
}
|
||||
|
||||
export interface CloudintegrationtypesServiceMetadataDTO {
|
||||
@@ -3882,6 +3915,10 @@ export interface CloudintegrationtypesPostableAgentCheckInDTO {
|
||||
* @type string
|
||||
*/
|
||||
providerAccountId?: string;
|
||||
/**
|
||||
* @type integer,null
|
||||
*/
|
||||
syncedVersion?: number | null;
|
||||
}
|
||||
|
||||
export interface CloudintegrationtypesStorableIntegrationDashboardDTO {
|
||||
|
||||
@@ -24,6 +24,7 @@ const accountsResponse: ListAccounts200 = {
|
||||
agentReport: {
|
||||
timestampMillis: 1747114366214,
|
||||
data: null,
|
||||
syncState: null,
|
||||
},
|
||||
providerAccountId: PROVIDER_ACCOUNT_ID,
|
||||
removedAt: null,
|
||||
|
||||
@@ -295,7 +295,11 @@ const account = (
|
||||
provider,
|
||||
providerAccountId: ACCOUNTS[provider][index],
|
||||
config: accountConfig(provider),
|
||||
agentReport: { timestampMillis: Date.now() - 45 * 1000, data: null },
|
||||
agentReport: {
|
||||
timestampMillis: Date.now() - 45 * 1000,
|
||||
data: null,
|
||||
syncState: null,
|
||||
},
|
||||
createdAt: new Date(Date.now() - 21 * 24 * 60 * 60 * 1000).toISOString(),
|
||||
updatedAt: new Date(Date.now() - 60 * 60 * 1000).toISOString(),
|
||||
removedAt: null,
|
||||
|
||||
@@ -134,6 +134,24 @@ func (store *store) UpdateAccount(ctx context.Context, account *cloudintegration
|
||||
BunDBCtx(ctx).
|
||||
NewUpdate().
|
||||
Model(account).
|
||||
Column("config").
|
||||
Column("updated_at").
|
||||
WherePK().
|
||||
Where("org_id = ?", account.OrgID).
|
||||
Where("provider = ?", account.Provider).
|
||||
Exec(ctx)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (store *store) UpdateAgentReport(ctx context.Context, account *cloudintegrationtypes.StorableCloudIntegration) error {
|
||||
_, err := store.
|
||||
store.
|
||||
BunDBCtx(ctx).
|
||||
NewUpdate().
|
||||
Model(account).
|
||||
Column("account_id").
|
||||
Column("last_agent_report").
|
||||
WherePK().
|
||||
Where("org_id = ?", account.OrgID).
|
||||
Where("provider = ?", account.Provider).
|
||||
|
||||
@@ -26,6 +26,17 @@ type Account struct {
|
||||
type AgentReport struct {
|
||||
TimestampMillis int64 `json:"timestampMillis" required:"true"`
|
||||
Data map[string]any `json:"data" required:"true" nullable:"true"`
|
||||
SyncState *SyncState `json:"syncState" required:"true" nullable:"true"`
|
||||
}
|
||||
|
||||
type SyncState struct {
|
||||
Version int64 `json:"version" required:"true"`
|
||||
InSync bool `json:"inSync" required:"true"`
|
||||
Regions map[string]*RegionSyncState `json:"regions" required:"true" nullable:"false"`
|
||||
}
|
||||
|
||||
type RegionSyncState struct {
|
||||
State RegionState `json:"state" required:"true"`
|
||||
}
|
||||
|
||||
type AccountConfig struct {
|
||||
@@ -150,6 +161,7 @@ func NewAccountFromStorable(storableAccount *StorableCloudIntegration) (*Account
|
||||
account.AgentReport = &AgentReport{
|
||||
TimestampMillis: storableAccount.LastAgentReport.TimestampMillis,
|
||||
Data: storableAccount.LastAgentReport.Data,
|
||||
SyncState: NewSyncStateFromStorable(storableAccount.LastAgentReport.SyncState),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -308,10 +320,101 @@ func NewAccountConfigFromUpdatable(provider CloudProviderType, config *Updatable
|
||||
}
|
||||
}
|
||||
|
||||
func NewAgentReport(data map[string]any) *AgentReport {
|
||||
func NewAgentReport(data map[string]any, syncState *SyncState) *AgentReport {
|
||||
return &AgentReport{
|
||||
TimestampMillis: time.Now().UnixMilli(),
|
||||
Data: data,
|
||||
SyncState: syncState,
|
||||
}
|
||||
}
|
||||
|
||||
// NewSyncState returns the sync state after a check-in without mutating previous.
|
||||
// The ack is applied before the config diff, so it is checked against the version the agent was last sent.
|
||||
func NewSyncState(previous *SyncState, regions []string, removed bool, syncedVersion *int64) *SyncState {
|
||||
next := &SyncState{Version: 1, InSync: true, Regions: make(map[string]*RegionSyncState)}
|
||||
|
||||
// First check-in: seed from the config as in sync. Otherwise start from a copy of previous.
|
||||
if previous == nil {
|
||||
for _, region := range regions {
|
||||
next.Regions[region] = &RegionSyncState{State: RegionStatePresent}
|
||||
}
|
||||
} else {
|
||||
next.Version = previous.Version
|
||||
next.InSync = previous.InSync
|
||||
for region, regionSyncState := range previous.Regions {
|
||||
next.Regions[region] = &RegionSyncState{State: regionSyncState.State}
|
||||
}
|
||||
}
|
||||
|
||||
// The agent synced this version, so its removed regions are cleaned up and can be dropped.
|
||||
if syncedVersion != nil && *syncedVersion == next.Version {
|
||||
next.InSync = true
|
||||
for region, regionSyncState := range next.Regions {
|
||||
if regionSyncState.State == RegionStateRemoved {
|
||||
delete(next.Regions, region)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
changed := false
|
||||
|
||||
if removed {
|
||||
// Integration removed: every present region must be cleaned up.
|
||||
for _, regionSyncState := range next.Regions {
|
||||
if regionSyncState.State != RegionStateRemoved {
|
||||
regionSyncState.State = RegionStateRemoved
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
} else {
|
||||
desiredRegions := make(map[string]struct{}, len(regions))
|
||||
for _, region := range regions {
|
||||
desiredRegions[region] = struct{}{}
|
||||
|
||||
regionSyncState, ok := next.Regions[region]
|
||||
switch {
|
||||
case !ok:
|
||||
// Region added to the config.
|
||||
next.Regions[region] = &RegionSyncState{State: RegionStatePresent}
|
||||
changed = true
|
||||
case regionSyncState.State == RegionStateRemoved:
|
||||
// Region added back before its removal was acked.
|
||||
regionSyncState.State = RegionStatePresent
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
|
||||
for region, regionSyncState := range next.Regions {
|
||||
if _, desired := desiredRegions[region]; !desired && regionSyncState.State == RegionStatePresent {
|
||||
// Region removed from the config.
|
||||
regionSyncState.State = RegionStateRemoved
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if changed {
|
||||
next.Version++
|
||||
next.InSync = false
|
||||
}
|
||||
|
||||
return next
|
||||
}
|
||||
|
||||
func NewSyncStateFromStorable(storableSyncState *StorableSyncState) *SyncState {
|
||||
if storableSyncState == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
regions := make(map[string]*RegionSyncState, len(storableSyncState.Regions))
|
||||
for region, regionSyncState := range storableSyncState.Regions {
|
||||
regions[region] = &RegionSyncState{State: regionSyncState.State}
|
||||
}
|
||||
|
||||
return &SyncState{
|
||||
Version: storableSyncState.Version,
|
||||
InSync: storableSyncState.InSync,
|
||||
Regions: regions,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -335,6 +438,26 @@ func (account *Account) Update(provider CloudProviderType, config *AccountConfig
|
||||
return nil
|
||||
}
|
||||
|
||||
// NextSyncState returns the sync state for this check-in, or nil for providers without one.
|
||||
func (account *Account) NextSyncState(syncedVersion *int64) *SyncState {
|
||||
if account.Provider != CloudProviderTypeAWS {
|
||||
return nil
|
||||
}
|
||||
|
||||
var previous *SyncState
|
||||
if account.AgentReport != nil {
|
||||
previous = account.AgentReport.SyncState
|
||||
}
|
||||
|
||||
regions := account.Config.AWS.Regions
|
||||
// Removed before the agent ever checked in: no region was sent to it, so there is nothing to clean up.
|
||||
if account.AgentReport == nil && account.RemovedAt != nil {
|
||||
regions = nil
|
||||
}
|
||||
|
||||
return NewSyncState(previous, regions, account.RemovedAt != nil, syncedVersion)
|
||||
}
|
||||
|
||||
func (postableAccount *PostableAccount) UnmarshalJSON(data []byte) error {
|
||||
type Alias PostableAccount
|
||||
|
||||
|
||||
@@ -12,7 +12,8 @@ type AgentCheckInRequest struct {
|
||||
ProviderAccountID string `json:"providerAccountId" required:"false"`
|
||||
CloudIntegrationID valuer.UUID `json:"cloudIntegrationId" required:"false"`
|
||||
|
||||
Data map[string]any `json:"data" required:"true" nullable:"true"`
|
||||
Data map[string]any `json:"data" required:"true" nullable:"true"`
|
||||
SyncedVersion *int64 `json:"syncedVersion" required:"false" nullable:"true"`
|
||||
}
|
||||
|
||||
type PostableAgentCheckIn struct {
|
||||
@@ -28,6 +29,7 @@ type AgentCheckInResponse struct {
|
||||
ProviderAccountID string `json:"providerAccountId" required:"true"`
|
||||
IntegrationConfig *ProviderIntegrationConfig `json:"integrationConfig" required:"true"`
|
||||
RemovedAt *time.Time `json:"removedAt" required:"true" nullable:"true"`
|
||||
SyncState *SyncState `json:"syncState" required:"true" nullable:"true"`
|
||||
}
|
||||
|
||||
type GettableAgentCheckIn struct {
|
||||
@@ -73,12 +75,13 @@ func NewGettableAgentCheckIn(provider CloudProviderType, resp *AgentCheckInRespo
|
||||
return gettable
|
||||
}
|
||||
|
||||
func NewAgentCheckInResponse(providerAccountID, cloudIntegrationID string, integrationConfig *ProviderIntegrationConfig, removedAt *time.Time) *AgentCheckInResponse {
|
||||
func NewAgentCheckInResponse(providerAccountID, cloudIntegrationID string, integrationConfig *ProviderIntegrationConfig, removedAt *time.Time, syncState *SyncState) *AgentCheckInResponse {
|
||||
return &AgentCheckInResponse{
|
||||
CloudIntegrationID: cloudIntegrationID,
|
||||
ProviderAccountID: providerAccountID,
|
||||
IntegrationConfig: integrationConfig,
|
||||
RemovedAt: removedAt,
|
||||
SyncState: syncState,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -25,6 +25,17 @@ var (
|
||||
ErrCodeServiceDefinitionNotFound = errors.MustNewCode("service_definition_not_found")
|
||||
)
|
||||
|
||||
var (
|
||||
RegionStatePresent = RegionState{valuer.NewString("present")}
|
||||
RegionStateRemoved = RegionState{valuer.NewString("removed")}
|
||||
)
|
||||
|
||||
type RegionState struct{ valuer.String }
|
||||
|
||||
func (RegionState) Enum() []any {
|
||||
return []any{RegionStatePresent, RegionStateRemoved}
|
||||
}
|
||||
|
||||
// StorableCloudIntegration represents a cloud integration stored in the database.
|
||||
// This is also referred as "Account" in the context of cloud integrations.
|
||||
type StorableCloudIntegration struct {
|
||||
@@ -43,8 +54,16 @@ type StorableCloudIntegration struct {
|
||||
// StorableAgentReport represents the last heartbeat and arbitrary data sent by the agent
|
||||
// as of now there is no use case for Data field, but keeping it for backwards compatibility with older structure.
|
||||
type StorableAgentReport struct {
|
||||
TimestampMillis int64 `json:"timestamp_millis"` // backward compatibility
|
||||
Data map[string]any `json:"data"`
|
||||
TimestampMillis int64 `json:"timestamp_millis"` // backward compatibility
|
||||
Data map[string]any `json:"data"`
|
||||
SyncState *StorableSyncState `json:"sync_state,omitempty"`
|
||||
}
|
||||
|
||||
// StorableSyncState holds every region sent to the agent. A removed region is dropped only after the agent acks Version.
|
||||
type StorableSyncState struct {
|
||||
Version int64 `json:"version"`
|
||||
InSync bool `json:"in_sync"`
|
||||
Regions map[string]*RegionSyncState `json:"regions"`
|
||||
}
|
||||
|
||||
// StorableCloudIntegrationService is to store service config for a cloud integration, which is a cloud provider specific configuration.
|
||||
@@ -148,12 +167,30 @@ func NewStorableCloudIntegration(account *Account) (*StorableCloudIntegration, e
|
||||
storableAccount.LastAgentReport = &StorableAgentReport{
|
||||
TimestampMillis: account.AgentReport.TimestampMillis,
|
||||
Data: account.AgentReport.Data,
|
||||
SyncState: NewStorableSyncState(account.AgentReport.SyncState),
|
||||
}
|
||||
}
|
||||
|
||||
return storableAccount, nil
|
||||
}
|
||||
|
||||
func NewStorableSyncState(syncState *SyncState) *StorableSyncState {
|
||||
if syncState == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
regions := make(map[string]*RegionSyncState, len(syncState.Regions))
|
||||
for region, regionSyncState := range syncState.Regions {
|
||||
regions[region] = &RegionSyncState{State: regionSyncState.State}
|
||||
}
|
||||
|
||||
return &StorableSyncState{
|
||||
Version: syncState.Version,
|
||||
InSync: syncState.InSync,
|
||||
Regions: regions,
|
||||
}
|
||||
}
|
||||
|
||||
// NewStorableCloudIntegrationService creates a new StorableCloudIntegrationService with
|
||||
// generated ID and timestamps from a CloudIntegrationService and its serialized config JSON.
|
||||
func NewStorableCloudIntegrationService(svc *CloudIntegrationService, configJSON string) *StorableCloudIntegrationService {
|
||||
@@ -172,6 +209,7 @@ func (account *StorableCloudIntegration) Update(providerAccountID *string, agent
|
||||
account.LastAgentReport = &StorableAgentReport{
|
||||
TimestampMillis: agentReport.TimestampMillis,
|
||||
Data: agentReport.Data,
|
||||
SyncState: NewStorableSyncState(agentReport.SyncState),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,9 +25,12 @@ type Store interface {
|
||||
// CreateAccount creates a new cloud integration account
|
||||
CreateAccount(ctx context.Context, account *StorableCloudIntegration) error
|
||||
|
||||
// UpdateAccount updates an existing cloud integration account
|
||||
// UpdateAccount updates the user updatable fields (config) of an existing cloud integration account
|
||||
UpdateAccount(ctx context.Context, account *StorableCloudIntegration) error
|
||||
|
||||
// UpdateAgentReport updates the provider account id and last agent report of an existing cloud integration account
|
||||
UpdateAgentReport(ctx context.Context, account *StorableCloudIntegration) error
|
||||
|
||||
// RemoveAccount marks a cloud integration account as removed by setting the RemovedAt field
|
||||
RemoveAccount(ctx context.Context, orgID, id valuer.UUID, provider CloudProviderType) error
|
||||
|
||||
|
||||
5
tests/fixtures/cloudintegrations.py
vendored
5
tests/fixtures/cloudintegrations.py
vendored
@@ -34,6 +34,8 @@ class ProviderAccountSpec:
|
||||
expected_config: Callable[[dict], dict]
|
||||
# only the suites that exercise updates need to supply it.
|
||||
updated_params: dict = field(default_factory=dict)
|
||||
# params -> the agentReport.syncState the API is expected to return after the first check-in.
|
||||
expected_sync_state: Callable[[dict], dict | None] = lambda p: None
|
||||
# id shown in parametrized test names; defaults to the provider slug.
|
||||
id: str = field(default="")
|
||||
|
||||
@@ -315,6 +317,7 @@ def simulate_agent_checkin(
|
||||
account_id: str,
|
||||
cloud_account_id: str,
|
||||
data: dict | None = None,
|
||||
synced_version: int | None = None,
|
||||
) -> requests.Response:
|
||||
endpoint = f"/api/v1/cloud_integrations/{cloud_provider}/accounts/check_in"
|
||||
|
||||
@@ -323,6 +326,8 @@ def simulate_agent_checkin(
|
||||
"providerAccountId": cloud_account_id,
|
||||
"data": data or {},
|
||||
}
|
||||
if synced_version is not None:
|
||||
checkin_payload["syncedVersion"] = synced_version
|
||||
|
||||
response = requests.post(
|
||||
signoz.self.host_configs["8080"].get(endpoint),
|
||||
|
||||
@@ -3,6 +3,7 @@ from collections.abc import Callable
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD, add_license
|
||||
@@ -152,3 +153,230 @@ def test_duplicate_cloud_account_checkins(
|
||||
# Second check-in: account2 tries to claim the same provider account ID → 409
|
||||
response = simulate_agent_checkin(signoz, admin_token, spec.provider, account2["id"], same_provider_account_id)
|
||||
assert response.status_code == HTTPStatus.CONFLICT, f"Expected 409 for duplicate providerAccountId, got {response.status_code}: {response.text}"
|
||||
|
||||
|
||||
def test_sync_state_drops_removed_region_after_ack(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
create_cloud_integration_account: Callable,
|
||||
) -> None:
|
||||
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
account_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
|
||||
provider_account_id = str(uuid.uuid4())
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
response = requests.put(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/aws/accounts/{account_id}"),
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
json={"config": {"aws": {"regions": ["us-east-1"]}}},
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.text
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 2,
|
||||
"inSync": False,
|
||||
"regions": {"us-east-1": {"state": "present"}, "us-west-2": {"state": "removed"}},
|
||||
}, "removed region should be marked removed and the version bumped"
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id, synced_version=2)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 2,
|
||||
"inSync": True,
|
||||
"regions": {"us-east-1": {"state": "present"}},
|
||||
}, "acked removed region should be dropped"
|
||||
|
||||
|
||||
def test_sync_state_keeps_removed_region_without_ack(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
create_cloud_integration_account: Callable,
|
||||
) -> None:
|
||||
"""The agent failed to clean up or crashed, so it never acks: the removed region stays and the version stays put."""
|
||||
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
account_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
|
||||
provider_account_id = str(uuid.uuid4())
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
response = requests.put(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/aws/accounts/{account_id}"),
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
json={"config": {"aws": {"regions": ["us-east-1"]}}},
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.text
|
||||
|
||||
for _ in range(3):
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 2,
|
||||
"inSync": False,
|
||||
"regions": {"us-east-1": {"state": "present"}, "us-west-2": {"state": "removed"}},
|
||||
}, "unacked removed region should stay without bumping the version"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("synced_version", [2, 9], ids=["stale", "ahead"])
|
||||
def test_sync_state_ignores_mismatched_ack(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
create_cloud_integration_account: Callable,
|
||||
synced_version: int,
|
||||
) -> None:
|
||||
"""An ack for any version other than the current one (v3) is ignored,
|
||||
so us-west-2, removed at v2 and still unacked, is not dropped.
|
||||
"""
|
||||
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
account_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
|
||||
provider_account_id = str(uuid.uuid4())
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
for regions in (["us-east-1"], ["us-east-1", "eu-west-1"]):
|
||||
response = requests.put(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/aws/accounts/{account_id}"),
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
json={"config": {"aws": {"regions": regions}}},
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.text
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
expected_sync_state = {
|
||||
"version": 3,
|
||||
"inSync": False,
|
||||
"regions": {"us-east-1": {"state": "present"}, "us-west-2": {"state": "removed"}, "eu-west-1": {"state": "present"}},
|
||||
}
|
||||
assert response.json()["data"]["syncState"] == expected_sync_state
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id, synced_version=synced_version)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == expected_sync_state, "an ack for another version should be ignored"
|
||||
|
||||
|
||||
def test_sync_state_applies_ack_before_config_change(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
create_cloud_integration_account: Callable,
|
||||
) -> None:
|
||||
"""The user changes regions while the agent syncs: the ack for the version it synced still lands."""
|
||||
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
account_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
|
||||
provider_account_id = str(uuid.uuid4())
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
for regions, synced_version in ((["us-east-1"], None), (["us-east-1", "eu-west-1"], 2)):
|
||||
response = requests.put(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/aws/accounts/{account_id}"),
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
json={"config": {"aws": {"regions": regions}}},
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.text
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id, synced_version=synced_version)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 3,
|
||||
"inSync": False,
|
||||
"regions": {"us-east-1": {"state": "present"}, "eu-west-1": {"state": "present"}},
|
||||
}, "ack should drop the removed region before the new region bumps the version"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("synced_version", [1, None], ids=["agent_acks_synced_version", "agent_crashed"])
|
||||
def test_sync_state_region_removed_during_sync(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
create_cloud_integration_account: Callable,
|
||||
synced_version: int | None,
|
||||
) -> None:
|
||||
"""The user removes a region while the agent syncs v1; whether the agent acks v1 or crashed, the region must not be lost."""
|
||||
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
account_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
|
||||
provider_account_id = str(uuid.uuid4())
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 1,
|
||||
"inSync": True,
|
||||
"regions": {"us-east-1": {"state": "present"}, "us-west-2": {"state": "present"}},
|
||||
}
|
||||
|
||||
response = requests.put(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/aws/accounts/{account_id}"),
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
json={"config": {"aws": {"regions": ["us-east-1"]}}},
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.text
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id, synced_version=synced_version)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 2,
|
||||
"inSync": False,
|
||||
"regions": {"us-east-1": {"state": "present"}, "us-west-2": {"state": "removed"}},
|
||||
}, "region removed mid-sync should be marked removed"
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id, synced_version=2)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 2,
|
||||
"inSync": True,
|
||||
"regions": {"us-east-1": {"state": "present"}},
|
||||
}
|
||||
|
||||
|
||||
def test_sync_state_after_disconnect(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
create_cloud_integration_account: Callable,
|
||||
) -> None:
|
||||
admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
account_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
|
||||
provider_account_id = str(uuid.uuid4())
|
||||
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
response = requests.delete(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/aws/accounts/{account_id}"),
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.text
|
||||
|
||||
for _ in range(2):
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["removedAt"] is not None, "removedAt should be set after disconnect"
|
||||
assert response.json()["data"]["syncState"] == {
|
||||
"version": 2,
|
||||
"inSync": False,
|
||||
"regions": {"us-east-1": {"state": "removed"}, "us-west-2": {"state": "removed"}},
|
||||
}, "every region should be removed once, without bumping the version on later check-ins"
|
||||
|
||||
for _ in range(2):
|
||||
response = simulate_agent_checkin(signoz, admin_token, "aws", account_id, provider_account_id, synced_version=2)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["data"]["syncState"] == {"version": 2, "inSync": True, "regions": {}}, "acked removal should leave no regions"
|
||||
|
||||
@@ -21,6 +21,11 @@ AWS_ACCOUNT_SPEC = ProviderAccountSpec(
|
||||
updated_params={"deployment_region": "us-east-1", "regions": ["us-east-1", "us-west-2", "eu-west-1"]},
|
||||
build_config=lambda p: {"aws": {"deploymentRegion": p["deployment_region"], "regions": p["regions"]}},
|
||||
expected_config=lambda p: {"regions": p["regions"]},
|
||||
expected_sync_state=lambda p: {
|
||||
"version": 1,
|
||||
"inSync": True,
|
||||
"regions": {region: {"state": "present"} for region in p["regions"]},
|
||||
},
|
||||
)
|
||||
|
||||
GCP_ACCOUNT_SPEC = ProviderAccountSpec(
|
||||
@@ -128,6 +133,7 @@ def test_list_accounts_after_checkin(
|
||||
assert found["providerAccountId"] == provider_account_id, "providerAccountId should match"
|
||||
assert found["config"][spec.provider] == spec.expected_config(spec.initial_params), "config should match account config"
|
||||
assert found["agentReport"] is not None, "agentReport should be present after check-in"
|
||||
assert found["agentReport"]["syncState"] == spec.expected_sync_state(spec.initial_params), "syncState should be seeded from the account regions on first check-in"
|
||||
assert found["removedAt"] is None, "removedAt should be null for a live account"
|
||||
|
||||
|
||||
@@ -282,6 +288,7 @@ def test_update_account_after_checkin_preserves_connected_status(
|
||||
assert found_after is not None, "Account must still be listed after config update (account_id should not be reset)"
|
||||
assert found_after["providerAccountId"] == provider_account_id, "providerAccountId should be preserved after update"
|
||||
assert found_after["agentReport"] is not None, "agentReport should be preserved after update"
|
||||
assert found_after["agentReport"]["syncState"] == found_before["agentReport"]["syncState"], "config update must not change syncState"
|
||||
assert found_after["config"][spec.provider] == spec.expected_config(spec.updated_params), "Config should reflect the update"
|
||||
assert found_after["removedAt"] is None, "removedAt should still be null"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user