Merge branch 'main' of https://github.com/SigNoz/signoz into keystone-pod/issues/148

#### Description

- Opamp places user log pipelines around the first
`signozlogsnormalizer` in the logs pipeline, per its `opamp.position`:
the normalizer runs after them by default (`after`), before them with
`before`. Other processors keep their places.
- Without the normalizer, processor ordering is unchanged.
- The `use_json_body`-prepended normalize operator is dropped when the
normalizer already runs ahead of user pipelines (`before`); otherwise
both are kept.
- Removes the `json_body_dual_ingestion` flag: the normalize operator no
longer stashes the original body, and isn't appended after user
pipelines when only that flag is on.

| `use_json_body` | `opamp.position` | Logs pipeline | Normalize
operator |
|---|---|---|---|
| off | no normalizer | memory_limiter → user → custom → batch
(unchanged) | — |
| off | `after` (default) | … → user → normalizer → … | — |
| off | `before` | … → normalizer → user → … | — |
| on | no normalizer | memory_limiter → normalize op → user → custom →
batch | prepended |
| on | `after` (default) | … → normalize op → user → normalizer → … |
prepended (both normalize) |
| on | `before` | … → normalizer → user → … | dropped |

#### Additional Information

- Depends on SigNoz/signoz-otel-collector#953 for the `opamp.position`
config.
- The pipelines preview still prepends the operator under
`use_json_body`, since it can't see per-collector `opamp.position`.
- Configs still setting `json_body_dual_ingestion` keep working; the
unknown flag is skipped with a warning.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

.claude/rules/pytest.md

@@ -17,3 +17,4 @@ For the Python integration suite under `tests/`. Setup, running, and suite layou
 - **snake_case parametrize ids.** `ids=["fill_gaps", "fill_zero"]`, not camelCase.
 - **Name suite files with the two-digit prefix (`NN_*.py`).** `pyproject.toml` restricts collection to `[0-9][0-9]_*.py` (plus the bootstrap `setup.py` / `run.py`) — a file that doesn't match is silently never collected.
 - **Always run pytest from `tests/`.** `--import-mode=importlib` is what allows same-basename files across suites (`querier/01_logs.py` vs `rawexportdata/01_logs.py`), but it disables pytest's implicit `sys.path` injection — `import fixtures` only resolves via `pythonpath = ["."]` from that rootdir.
+- **After changing any `.py` file, run `make py-fmt` and `make py-lint`** from the repo root and fix what they report.

docs/api/openapi.yml

@@ -2159,6 +2159,8 @@ components:
       type: object
     CloudintegrationtypesGCPAccountConfig:
       properties:
+        connectionMethod:
+          $ref: '#/components/schemas/CloudintegrationtypesGCPConnectionMethod'
         deploymentProjectId:
           type: string
         deploymentRegion:
@@ -2168,13 +2170,83 @@ components:
             type: string
           type: array
       required:
+      - connectionMethod
       - deploymentProjectId
       - deploymentRegion
       - projectIds
       type: object
     CloudintegrationtypesGCPConnectionArtifact:
+      properties:
+        cliCommand:
+          type: string
+        createSecretsCommand:
+          type: string
+        createServiceAccountCommand:
+          type: string
+      required:
+      - cliCommand
+      - createSecretsCommand
+      - createServiceAccountCommand
       type: object
+    CloudintegrationtypesGCPConnectionMethod:
+      enum:
+      - manual
+      - agent
+      type: string
     CloudintegrationtypesGCPIntegrationConfig:
+      properties:
+        connectionMethod:
+          $ref: '#/components/schemas/CloudintegrationtypesGCPConnectionMethod'
+        deploymentProjectId:
+          type: string
+        deploymentRegion:
+          type: string
+        projectIds:
+          items:
+            type: string
+          type: array
+        telemetryCollectionStrategy:
+          $ref: '#/components/schemas/CloudintegrationtypesGCPTelemetryCollectionStrategy'
+      required:
+      - connectionMethod
+      - deploymentProjectId
+      - deploymentRegion
+      - projectIds
+      - telemetryCollectionStrategy
+      type: object
+    CloudintegrationtypesGCPLogsCollectionStrategy:
+      properties:
+        sinkFilter:
+          type: string
+      required:
+      - sinkFilter
+      type: object
+    CloudintegrationtypesGCPMetricsCollectionStrategy:
+      properties:
+        metricNames:
+          items:
+            type: string
+          type: array
+      required:
+      - metricNames
+      type: object
+    CloudintegrationtypesGCPPostableAccountConfig:
+      properties:
+        connectionMethod:
+          $ref: '#/components/schemas/CloudintegrationtypesGCPConnectionMethod'
+        deploymentProjectId:
+          type: string
+        deploymentRegion:
+          type: string
+        projectIds:
+          items:
+            type: string
+          type: array
+      required:
+      - connectionMethod
+      - deploymentProjectId
+      - deploymentRegion
+      - projectIds
       type: object
     CloudintegrationtypesGCPServiceConfig:
       properties:
@@ -2197,6 +2269,13 @@ components:
       required:
       - enabled
       type: object
+    CloudintegrationtypesGCPTelemetryCollectionStrategy:
+      properties:
+        logs:
+          $ref: '#/components/schemas/CloudintegrationtypesGCPLogsCollectionStrategy'
+        metrics:
+          $ref: '#/components/schemas/CloudintegrationtypesGCPMetricsCollectionStrategy'
+      type: object
     CloudintegrationtypesGettableAccountWithConnectionArtifact:
       properties:
         connectionArtifact:
@@ -2333,7 +2412,7 @@ components:
         azure:
           $ref: '#/components/schemas/CloudintegrationtypesAzureAccountConfig'
         gcp:
-          $ref: '#/components/schemas/CloudintegrationtypesGCPAccountConfig'
+          $ref: '#/components/schemas/CloudintegrationtypesGCPPostableAccountConfig'
       type: object
     CloudintegrationtypesPostableAgentCheckIn:
       properties:
@@ -2568,7 +2647,6 @@ components:
         projectIds:
           items:
             type: string
-          nullable: true
           type: array
       required:
       - deploymentProjectId

ee/modules/cloudintegration/implcloudintegration/implcloudprovider/gcpcloudprovider.go

@@ -7,24 +7,66 @@ import (
 	"github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
 )
 
-type gcpcloudprovider struct {
-	serviceDefinitions cloudintegrationtypes.ServiceDefinitionStore
-}
-
 func NewGCPCloudProvider(defStore cloudintegrationtypes.ServiceDefinitionStore) cloudintegration.CloudProviderModule {
 	return &gcpcloudprovider{
 		serviceDefinitions: defStore,
 	}
 }
 
-func (g *gcpcloudprovider) BuildIntegrationConfig(ctx context.Context, account *cloudintegrationtypes.Account, services []*cloudintegrationtypes.StorableCloudIntegrationService) (*cloudintegrationtypes.ProviderIntegrationConfig, error) {
-	// for manual flow we don't have any integration config to return, so returning empty config for now.
-	return &cloudintegrationtypes.ProviderIntegrationConfig{}, nil
+func (g *gcpcloudprovider) BuildIntegrationConfig(
+	ctx context.Context,
+	account *cloudintegrationtypes.Account,
+	services []*cloudintegrationtypes.StorableCloudIntegrationService,
+) (*cloudintegrationtypes.ProviderIntegrationConfig, error) {
+	if account.Config.GCP.ConnectionMethod != cloudintegrationtypes.GCPConnectionMethodAgent {
+		return &cloudintegrationtypes.ProviderIntegrationConfig{}, nil
+	}
+
+	var metricNames []string
+	var sinkFilters []string
+
+	for _, storedSvc := range services {
+		svcCfg, err := cloudintegrationtypes.NewServiceConfigFromJSON(cloudintegrationtypes.CloudProviderTypeGCP, storedSvc.Config)
+		if err != nil {
+			return nil, err
+		}
+
+		svcDef, err := g.GetServiceDefinition(ctx, storedSvc.Type)
+		if err != nil {
+			return nil, err
+		}
+
+		if svcDef.TelemetryCollectionStrategy.GCP == nil {
+			continue
+		}
+
+		strategy := svcDef.TelemetryCollectionStrategy.GCP
+		if svcCfg.IsMetricsEnabled(cloudintegrationtypes.CloudProviderTypeGCP) && strategy.Metrics != nil {
+			metricNames = append(metricNames, strategy.Metrics.MetricNames...)
+		}
+
+		if svcCfg.IsLogsEnabled(cloudintegrationtypes.CloudProviderTypeGCP) && strategy.Logs != nil {
+			sinkFilters = append(sinkFilters, strategy.Logs.SinkFilter)
+		}
+	}
+
+	return &cloudintegrationtypes.ProviderIntegrationConfig{
+		GCP: cloudintegrationtypes.NewGCPIntegrationConfig(account.Config.GCP, metricNames, sinkFilters),
+	}, nil
 }
 
 func (g *gcpcloudprovider) GetConnectionArtifact(ctx context.Context, account *cloudintegrationtypes.Account, req *cloudintegrationtypes.GetConnectionArtifactRequest) (*cloudintegrationtypes.ConnectionArtifact, error) {
-	// for manual flow we don't have any connection artifact to return, so returning empty artifact for now.
-	return &cloudintegrationtypes.ConnectionArtifact{}, nil
+	if account.Config.GCP.ConnectionMethod != cloudintegrationtypes.GCPConnectionMethodAgent {
+		return &cloudintegrationtypes.ConnectionArtifact{}, nil
+	}
+
+	connectionArtifact, err := cloudintegrationtypes.NewGCPConnectionArtifact(account.ID, req.Config.AgentVersion, req.Credentials, account.Config.GCP)
+	if err != nil {
+		return nil, err
+	}
+	return &cloudintegrationtypes.ConnectionArtifact{
+		GCP: connectionArtifact,
+	}, nil
 }
 
 func (g *gcpcloudprovider) GetServiceDefinition(ctx context.Context, serviceID cloudintegrationtypes.ServiceID) (*cloudintegrationtypes.ServiceDefinition, error) {
@@ -34,3 +76,7 @@ func (g *gcpcloudprovider) GetServiceDefinition(ctx context.Context, serviceID c
 func (g *gcpcloudprovider) ListServiceDefinitions(ctx context.Context) ([]*cloudintegrationtypes.ServiceDefinition, error) {
 	return g.serviceDefinitions.List(ctx, cloudintegrationtypes.CloudProviderTypeGCP)
 }
+
+type gcpcloudprovider struct {
+	serviceDefinitions cloudintegrationtypes.ServiceDefinitionStore
+}

ee/modules/cloudintegration/implcloudintegration/module.go

@@ -2,7 +2,6 @@ package implcloudintegration
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -171,18 +170,6 @@ func (module *module) AgentCheckIn(ctx context.Context, orgID valuer.UUID, provi
 		return nil, errors.New(errors.TypeLicenseUnavailable, errors.CodeLicenseUnavailable, "a valid license is not available").WithAdditional("this feature requires a valid license").WithAdditional(err.Error())
 	}
 
-	connectedAccount, err := module.store.GetConnectedAccountByProviderAccountID(ctx, orgID, req.ProviderAccountID, provider)
-	if err != nil && !errors.Ast(err, errors.TypeNotFound) {
-		return nil, err
-	}
-
-	// If a different integration is already connected to this provider account ID, reject the check-in.
-	// Allow re-check-in from the same integration (e.g. agent restarting).
-	if connectedAccount != nil && connectedAccount.ID != req.CloudIntegrationID {
-		errMessage := fmt.Sprintf("provider account id %s is already connected to cloud integration id %s", req.ProviderAccountID, connectedAccount.ID)
-		return nil, errors.New(errors.TypeAlreadyExists, cloudintegrationtypes.ErrCodeCloudIntegrationAlreadyConnected, errMessage)
-	}
-
 	storableAccount, err := module.store.GetAccountByID(ctx, orgID, req.CloudIntegrationID, provider)
 	if err != nil {
 		return nil, err
@@ -195,41 +182,36 @@ func (module *module) AgentCheckIn(ctx context.Context, orgID valuer.UUID, provi
 
 	syncState := account.NextSyncState(req.SyncedVersion)
 
-	// If account has been removed (disconnected), return a minimal response with empty integration config.
 	if account.RemovedAt != nil {
-		// Heartbeat stays frozen after removal, only the sync state is updated.
-		if account.AgentReport != nil && syncState != nil {
-			account.UpdateSyncState(syncState)
+		return module.checkInRemovedAccount(ctx, account, req.ProviderAccountID, syncState)
+	}
 
-			storableAccount, err = cloudintegrationtypes.NewStorableCloudIntegration(account)
-			if err != nil {
-				return nil, err
-			}
+	if err := account.ErrIfInvalidProviderAccountID(req.ProviderAccountID); err != nil {
+		return nil, err
+	}
 
-			err = module.store.UpdateAgentReport(ctx, storableAccount)
-			if err != nil {
-				return nil, err
-			}
-		}
+	connectedAccount, err := module.store.GetConnectedAccountByProviderAccountID(ctx, orgID, req.ProviderAccountID, provider)
+	if err != nil && !errors.Ast(err, errors.TypeNotFound) {
+		return nil, err
+	}
 
-		return cloudintegrationtypes.NewAgentCheckInResponse(
-			req.ProviderAccountID,
-			account.ID.StringValue(),
-			new(cloudintegrationtypes.ProviderIntegrationConfig),
-			account.RemovedAt,
-			syncState,
-		), nil
+	// If a different integration is already connected to this provider account ID, reject the check-in.
+	// Allow re-check-in from the same integration (e.g. agent restarting).
+	if connectedAccount != nil && connectedAccount.ID != req.CloudIntegrationID {
+		return nil, errors.Newf(
+			errors.TypeAlreadyExists,
+			cloudintegrationtypes.ErrCodeCloudIntegrationAlreadyConnected,
+			"provider account id %s is already connected to cloud integration id %s", req.ProviderAccountID, connectedAccount.ID,
+		)
 	}
 
 	account.UpdateAgentReport(&req.ProviderAccountID, cloudintegrationtypes.NewAgentReport(req.Data, syncState))
-
 	storableAccount, err = cloudintegrationtypes.NewStorableCloudIntegration(account)
 	if err != nil {
 		return nil, err
 	}
 
-	err = module.store.UpdateAgentReport(ctx, storableAccount)
-	if err != nil {
+	if err := module.store.UpdateAgentReport(ctx, storableAccount); err != nil {
 		return nil, err
 	}
 
@@ -498,11 +480,45 @@ func (module *module) Collect(ctx context.Context, orgID valuer.UUID) (map[strin
 		stats["cloudintegration.azure.connectedaccounts.count"] = azureAccountsCount
 	}
 
-	// NOTE: not adding stats for services for now.
+	gcpAccountsCount, err := module.store.CountConnectedAccounts(ctx, orgID, cloudintegrationtypes.CloudProviderTypeGCP)
+	if err == nil {
+		stats["cloudintegration.gcp.connectedaccounts.count"] = gcpAccountsCount
+	}
 
 	return stats, nil
 }
 
+// checkInRemovedAccount returns an empty integration config and saves cleanup acknowledgments without refreshing the heartbeat.
+func (module *module) checkInRemovedAccount(ctx context.Context, account *cloudintegrationtypes.Account, providerAccountID string, syncState *cloudintegrationtypes.SyncState) (*cloudintegrationtypes.AgentCheckInResponse, error) {
+	response := cloudintegrationtypes.NewAgentCheckInResponse(
+		providerAccountID,
+		account.ID.StringValue(),
+		new(cloudintegrationtypes.ProviderIntegrationConfig),
+		account.RemovedAt,
+		syncState,
+	)
+
+	if account.AgentReport == nil {
+		return response, nil
+	}
+
+	if syncState == nil {
+		return response, nil
+	}
+
+	account.UpdateSyncState(syncState)
+	storableAccount, err := cloudintegrationtypes.NewStorableCloudIntegration(account)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := module.store.UpdateAgentReport(ctx, storableAccount); err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
 func (module *module) getCloudProvider(provider cloudintegrationtypes.CloudProviderType) (cloudintegration.CloudProviderModule, error) {
 	if cloudProviderModule, ok := module.cloudProvidersMap[provider]; ok {
 		return cloudProviderModule, nil

frontend/src/api/generated/services/sigNoz.schemas.ts

@@ -3722,7 +3722,12 @@ export interface CloudintegrationtypesAzureAccountConfigDTO {
 	resourceGroups: string[];
 }
 
+export enum CloudintegrationtypesGCPConnectionMethodDTO {
+	manual = 'manual',
+	agent = 'agent',
+}
 export interface CloudintegrationtypesGCPAccountConfigDTO {
+	connectionMethod: CloudintegrationtypesGCPConnectionMethodDTO;
 	/**
 	 * @type string
 	 */
@@ -3969,7 +3974,18 @@ export interface CloudintegrationtypesCollectedMetricDTO {
 }
 
 export interface CloudintegrationtypesGCPConnectionArtifactDTO {
-	[key: string]: unknown;
+	/**
+	 * @type string
+	 */
+	cliCommand: string;
+	/**
+	 * @type string
+	 */
+	createSecretsCommand: string;
+	/**
+	 * @type string
+	 */
+	createServiceAccountCommand: string;
 }
 
 export interface CloudintegrationtypesConnectionArtifactDTO {
@@ -4008,8 +4024,56 @@ export interface CloudintegrationtypesDataCollectedDTO {
 	metrics?: CloudintegrationtypesCollectedMetricDTO[] | null;
 }
 
+export interface CloudintegrationtypesGCPLogsCollectionStrategyDTO {
+	/**
+	 * @type string
+	 */
+	sinkFilter: string;
+}
+
+export interface CloudintegrationtypesGCPMetricsCollectionStrategyDTO {
+	/**
+	 * @type array
+	 */
+	metricNames: string[];
+}
+
+export interface CloudintegrationtypesGCPTelemetryCollectionStrategyDTO {
+	logs?: CloudintegrationtypesGCPLogsCollectionStrategyDTO;
+	metrics?: CloudintegrationtypesGCPMetricsCollectionStrategyDTO;
+}
+
 export interface CloudintegrationtypesGCPIntegrationConfigDTO {
-	[key: string]: unknown;
+	connectionMethod: CloudintegrationtypesGCPConnectionMethodDTO;
+	/**
+	 * @type string
+	 */
+	deploymentProjectId: string;
+	/**
+	 * @type string
+	 */
+	deploymentRegion: string;
+	/**
+	 * @type array
+	 */
+	projectIds: string[];
+	telemetryCollectionStrategy: CloudintegrationtypesGCPTelemetryCollectionStrategyDTO;
+}
+
+export interface CloudintegrationtypesGCPPostableAccountConfigDTO {
+	connectionMethod: CloudintegrationtypesGCPConnectionMethodDTO;
+	/**
+	 * @type string
+	 */
+	deploymentProjectId: string;
+	/**
+	 * @type string
+	 */
+	deploymentRegion: string;
+	/**
+	 * @type array
+	 */
+	projectIds: string[];
 }
 
 export interface CloudintegrationtypesGettableAccountWithConnectionArtifactDTO {
@@ -4164,7 +4228,7 @@ export interface CloudintegrationtypesGettableServicesMetadataDTO {
 export interface CloudintegrationtypesPostableAccountConfigDTO {
 	aws?: CloudintegrationtypesAWSPostableAccountConfigDTO;
 	azure?: CloudintegrationtypesAzureAccountConfigDTO;
-	gcp?: CloudintegrationtypesGCPAccountConfigDTO;
+	gcp?: CloudintegrationtypesGCPPostableAccountConfigDTO;
 }
 
 export interface CloudintegrationtypesPostableAccountDTO {
@@ -4308,9 +4372,9 @@ export interface CloudintegrationtypesUpdatableGCPAccountConfigDTO {
 	 */
 	deploymentRegion: string;
 	/**
-	 * @type array,null
+	 * @type array
 	 */
-	projectIds: string[] | null;
+	projectIds: string[];
 }
 
 export interface CloudintegrationtypesUpdatableAccountConfigDTO {

frontend/src/container/Integrations/CloudIntegration/GoogleCloudPlatform/__tests__/CloudAccountSetupDrawer.test.tsx

@@ -19,7 +19,7 @@ import {
 
 // `useCloudAccountSetupDrawer` imports logEvent by relative path, which the
 // jest.config moduleNameMapper (keyed on the `api/common/logEvent` alias) does
-// not intercept — so mock the resolved module directly.
+// not intercept, so mock the resolved module directly.
 jest.mock('../../../../../api/common/logEvent', () => ({
 	__esModule: true,
 	default: jest.fn(),
@@ -49,7 +49,7 @@ describe('GCP CloudAccountSetupDrawer', () => {
 			rest.get(GCP_CREDENTIALS_URL, (_req, res, ctx) =>
 				res(ctx.status(200), ctx.json(connectionCredentialsResponse)),
 			),
-			// check_in is registered first — it is a more specific path than
+			// check_in is registered first, it is a more specific path than
 			// /accounts and msw matches handlers in registration order.
 			rest.post(GCP_CHECK_IN_URL, async (req: RestRequest, res, ctx) => {
 				checkInPayload = await req.json();
@@ -142,6 +142,7 @@ describe('GCP CloudAccountSetupDrawer', () => {
 		expect(createAccountPayload).toStrictEqual({
 			config: {
 				gcp: {
+					connectionMethod: 'manual',
 					deploymentRegion: 'asia-south1',
 					deploymentProjectId: 'my-deployment-project-123',
 					projectIds: ['project-a', 'project-b'],

frontend/src/hooks/integration/gcp/useCloudAccountSetupDrawer.ts

@@ -11,6 +11,7 @@ import {
 } from 'api/generated/services/cloudintegration';
 import {
 	CloudintegrationtypesCredentialsDTO,
+	CloudintegrationtypesGCPConnectionMethodDTO,
 	CloudintegrationtypesPostableAccountDTO,
 	RenderErrorResponseDTO,
 } from 'api/generated/services/sigNoz.schemas';
@@ -96,13 +97,14 @@ export function useCloudAccountSetupDrawer({
 				const payload: CloudintegrationtypesPostableAccountDTO = {
 					config: {
 						gcp: {
+							connectionMethod: CloudintegrationtypesGCPConnectionMethodDTO.manual,
 							deploymentRegion: values.deploymentRegion,
 							deploymentProjectId: values.deploymentProjectId,
 							projectIds: values.projectIds || [],
 						},
 					},
 					credentials: {
-						// Cloud users can't edit these — the backend-provided credentials are
+						// Cloud users can't edit these, the backend-provided credentials are
 						// authoritative. Enterprise users have no backend defaults and enter
 						// their own (validated non-empty), so their form values are used.
 						ingestionUrl: connectionParams?.data?.ingestionUrl || values.ingestionUrl,

frontend/src/pages/IntegrationsDetailsPage/CloudIntegration/stories/__story_mockdata__/cloudIntegration.ts

@@ -3,6 +3,7 @@
  * Do not hand-edit: regenerate instead.
  */
 
+import { CloudintegrationtypesGCPConnectionMethodDTO } from 'api/generated/services/sigNoz.schemas';
 import type {
 	CloudintegrationtypesAccountConfigDTO,
 	CloudintegrationtypesAccountDTO,
@@ -282,6 +283,7 @@ const accountConfig = (
 
 	return {
 		gcp: {
+			connectionMethod: CloudintegrationtypesGCPConnectionMethodDTO.manual,
 			deploymentRegion: 'us-central1',
 			deploymentProjectId: 'signoz-prod-4471',
 			projectIds: ['signoz-prod-4471', 'signoz-analytics-8820'],

pkg/flagger/registry.go

@@ -9,7 +9,6 @@ var (
 	FeaturePutMetersInZeus        = featuretypes.MustNewName("put_meters_in_zeus")
 	FeatureUseMeterReporter       = featuretypes.MustNewName("use_meter_reporter")
 	FeatureUseJSONBody            = featuretypes.MustNewName("use_json_body")
-	FeatureJSONBodyDualIngestion  = featuretypes.MustNewName("json_body_dual_ingestion")
 	FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
 	FeatureResolveSemconvFamilies = featuretypes.MustNewName("resolve_semconv_families")
 	FeatureUseTraceAttributesJSON = featuretypes.MustNewName("use_trace_attributes_json")
@@ -65,14 +64,6 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
-		&featuretypes.Feature{
-			Name:           FeatureJSONBodyDualIngestion,
-			Kind:           featuretypes.KindBoolean,
-			Stage:          featuretypes.StageExperimental,
-			Description:    "Controls whether the collector's normalize operator keeps the original log body so it is ingested into both the legacy body and the JSON body columns",
-			DefaultVariant: featuretypes.MustNewName("disabled"),
-			Variants:       featuretypes.NewBooleanVariants(),
-		},
 		&featuretypes.Feature{
 			Name:           FeatureEnableMetricsReduction,
 			Kind:           featuretypes.KindBoolean,

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudsql_mysql/integration.json

@@ -121,7 +121,33 @@
     "logs": []
   },
   "telemetryCollectionStrategy": {
-    "gcp": {}
+    "gcp": {
+      "metrics": {
+        "metricNames": [
+          "cloudsql.googleapis.com/database/up",
+          "cloudsql.googleapis.com/database/instance_state",
+          "cloudsql.googleapis.com/database/replication/replica_lag",
+          "cloudsql.googleapis.com/database/cpu/utilization",
+          "cloudsql.googleapis.com/database/memory/utilization",
+          "cloudsql.googleapis.com/database/disk/utilization",
+          "cloudsql.googleapis.com/database/network/connections",
+          "cloudsql.googleapis.com/database/mysql/queries",
+          "cloudsql.googleapis.com/database/mysql/dml_operations_count",
+          "cloudsql.googleapis.com/database/mysql/threads",
+          "cloudsql.googleapis.com/database/mysql/innodb/buffer_pool_reads_count",
+          "cloudsql.googleapis.com/database/mysql/innodb/buffer_pool_read_requests_count",
+          "cloudsql.googleapis.com/database/mysql/slow_queries_count",
+          "cloudsql.googleapis.com/database/mysql/aborted_connects_count",
+          "cloudsql.googleapis.com/database/mysql/innodb/deadlocks_count",
+          "cloudsql.googleapis.com/database/mysql/innodb/row_lock_waits_count",
+          "cloudsql.googleapis.com/database/disk/read_ops_count",
+          "cloudsql.googleapis.com/database/disk/write_ops_count"
+        ]
+      },
+      "logs": {
+        "sinkFilter": "resource.type=\"cloudsql_database\" AND (logName:\"/logs/cloudsql.googleapis.com%2Fmysql\" OR (log_id(\"cloudaudit.googleapis.com/data_access\") AND protoPayload.request.@type=\"type.googleapis.com/google.cloud.sql.audit.v1.MysqlAuditEntry\"))"
+      }
+    }
   },
   "assets": {
     "dashboards": [

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudsql_postgres/integration.json

@@ -5,7 +5,7 @@
   "overview": "file://overview.md",
   "supportedSignals": {
     "metrics": true,
-    "logs": false
+    "logs": true
   },
   "dataCollected": {
     "metrics": [
@@ -75,6 +75,12 @@
         "type": "Gauge",
         "description": ""
       },
+      {
+        "name": "cloudsql.googleapis.com/database/postgresql/insights/aggregate/io_time",
+        "unit": "Microseconds",
+        "type": "Sum",
+        "description": ""
+      },
       {
         "name": "cloudsql.googleapis.com/database/postgresql/insights/perquery/execution_time",
         "unit": "Microseconds",
@@ -91,7 +97,29 @@
     "logs": []
   },
   "telemetryCollectionStrategy": {
-    "gcp": {}
+    "gcp": {
+      "metrics": {
+        "metricNames": [
+          "cloudsql.googleapis.com/database/up",
+          "cloudsql.googleapis.com/database/cpu/utilization",
+          "cloudsql.googleapis.com/database/memory/utilization",
+          "cloudsql.googleapis.com/database/memory/usage",
+          "cloudsql.googleapis.com/database/disk/bytes_used",
+          "cloudsql.googleapis.com/database/postgresql/num_backends",
+          "cloudsql.googleapis.com/database/postgresql/num_backends_by_state",
+          "cloudsql.googleapis.com/database/postgresql/transaction_count",
+          "cloudsql.googleapis.com/database/postgresql/deadlock_count",
+          "cloudsql.googleapis.com/database/postgresql/vacuum/oldest_transaction_age",
+          "cloudsql.googleapis.com/database/postgresql/insights/aggregate/execution_time",
+          "cloudsql.googleapis.com/database/postgresql/insights/aggregate/io_time",
+          "cloudsql.googleapis.com/database/postgresql/insights/perquery/execution_time",
+          "cloudsql.googleapis.com/database/postgresql/replication/replica_byte_lag"
+        ]
+      },
+      "logs": {
+        "sinkFilter": "resource.type=\"cloudsql_database\" AND (logName:\"/logs/cloudsql.googleapis.com%2Fpostgres\" OR (log_id(\"cloudaudit.googleapis.com/data_access\") AND protoPayload.request.@type=\"type.googleapis.com/google.cloud.sql.audit.v1.PgAuditEntry\"))"
+      }
+    }
   },
   "assets": {
     "dashboards": [

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/cloudstorage/integration.json

@@ -5,7 +5,7 @@
   "overview": "file://overview.md",
   "supportedSignals": {
     "metrics": true,
-    "logs": false
+    "logs": true
   },
   "dataCollected": {
     "metrics": [
@@ -55,7 +55,22 @@
     "logs": []
   },
   "telemetryCollectionStrategy": {
-    "gcp": {}
+    "gcp": {
+      "metrics": {
+        "metricNames": [
+          "storage.googleapis.com/storage/v2/total_bytes",
+          "storage.googleapis.com/storage/v2/total_count",
+          "storage.googleapis.com/storage/v2/deleted_bytes",
+          "storage.googleapis.com/api/request_count",
+          "storage.googleapis.com/network/received_bytes_count",
+          "storage.googleapis.com/network/sent_bytes_count",
+          "storage.googleapis.com/authn/authentication_count"
+        ]
+      },
+      "logs": {
+        "sinkFilter": "resource.type=\"gcs_bucket\""
+      }
+    }
   },
   "assets": {
     "dashboards": [

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/computeengine/integration.json

@@ -5,7 +5,7 @@
   "overview": "file://overview.md",
   "supportedSignals": {
     "metrics": true,
-    "logs": false
+    "logs": true
   },
   "dataCollected": {
     "metrics": [
@@ -109,7 +109,31 @@
     "logs": []
   },
   "telemetryCollectionStrategy": {
-    "gcp": {}
+    "gcp": {
+      "metrics": {
+        "metricNames": [
+          "compute.googleapis.com/instance/uptime_total",
+          "compute.googleapis.com/instance/cpu/utilization",
+          "compute.googleapis.com/guest/memory/bytes_used",
+          "compute.googleapis.com/guest/memory/percent_used",
+          "agent.googleapis.com/memory/percent_used",
+          "compute.googleapis.com/instance/disk/average_io_latency",
+          "compute.googleapis.com/instance/disk/read_bytes_count",
+          "compute.googleapis.com/instance/disk/write_bytes_count",
+          "compute.googleapis.com/instance/disk/read_ops_count",
+          "compute.googleapis.com/instance/disk/write_ops_count",
+          "compute.googleapis.com/instance/disk/performance_status",
+          "compute.googleapis.com/instance/network/received_bytes_count",
+          "compute.googleapis.com/instance/network/sent_bytes_count",
+          "compute.googleapis.com/instance/network/received_packets_count",
+          "compute.googleapis.com/instance/network/sent_packets_count",
+          "compute.googleapis.com/firewall/dropped_packets_count"
+        ]
+      },
+      "logs": {
+        "sinkFilter": "resource.type=\"gce_instance\" OR protoPayload.serviceName=\"compute.googleapis.com\""
+      }
+    }
   },
   "assets": {
     "dashboards": [

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/gke/integration.json

@@ -5,7 +5,7 @@
   "overview": "file://overview.md",
   "supportedSignals": {
     "metrics": true,
-    "logs": false
+    "logs": true
   },
   "dataCollected": {
     "metrics": [
@@ -109,7 +109,31 @@
     "logs": []
   },
   "telemetryCollectionStrategy": {
-    "gcp": {}
+    "gcp": {
+      "metrics": {
+        "metricNames": [
+          "kubernetes.io/container/cpu/limit_utilization",
+          "kubernetes.io/container/cpu/request_utilization",
+          "kubernetes.io/container/memory/limit_utilization",
+          "kubernetes.io/container/memory/request_utilization",
+          "kubernetes.io/container/restart_count",
+          "kubernetes.io/pod/latencies/pod_first_ready",
+          "kubernetes.io/pod/network/received_bytes_count",
+          "kubernetes.io/pod/network/sent_bytes_count",
+          "kubernetes.io/pod/volume/utilization",
+          "kubernetes.io/node/cpu/allocatable_utilization",
+          "kubernetes.io/node/memory/allocatable_utilization",
+          "kubernetes.io/node/ephemeral_storage/used_bytes",
+          "kubernetes.io/node/ephemeral_storage/allocatable_bytes",
+          "kubernetes.io/node/network/received_bytes_count",
+          "kubernetes.io/node/network/sent_bytes_count",
+          "kubernetes.io/node/status_condition"
+        ]
+      },
+      "logs": {
+        "sinkFilter": "resource.type=(\"k8s_container\" OR \"k8s_pod\" OR \"k8s_node\" OR \"k8s_cluster\" OR \"k8s_control_plane_component\" OR \"gke_cluster\" OR \"gke_nodepool\")"
+      }
+    }
   },
   "assets": {
     "dashboards": [

pkg/modules/cloudintegration/implcloudintegration/fs/definitions/gcp/memorystore_redis/integration.json

@@ -5,7 +5,7 @@
   "overview": "file://overview.md",
   "supportedSignals": {
     "metrics": true,
-    "logs": false
+    "logs": true
   },
   "dataCollected": {
     "metrics": [
@@ -79,7 +79,26 @@
     "logs": []
   },
   "telemetryCollectionStrategy": {
-    "gcp": {}
+    "gcp": {
+      "metrics": {
+        "metricNames": [
+          "redis.googleapis.com/server/uptime",
+          "redis.googleapis.com/stats/memory/usage_ratio",
+          "redis.googleapis.com/stats/memory/system_memory_usage_ratio",
+          "redis.googleapis.com/stats/cpu_utilization_main_thread",
+          "redis.googleapis.com/clients/connected",
+          "redis.googleapis.com/stats/cache_hit_ratio",
+          "redis.googleapis.com/keyspace/keys",
+          "redis.googleapis.com/commands/usec_per_call",
+          "redis.googleapis.com/commands/calls",
+          "redis.googleapis.com/clients/blocked",
+          "redis.googleapis.com/stats/reject_connections_count"
+        ]
+      },
+      "logs": {
+        "sinkFilter": "resource.type=\"redis_instance\" OR (protoPayload.serviceName=\"redis.googleapis.com\" AND protoPayload.methodName:\".CloudRedis.\") OR (log_id(\"maintenance.googleapis.com/activity_log\") AND labels.activity_type_name:\"CloudRedisRollout\")"
+      }
+    }
   },
   "assets": {
     "dashboards": [

pkg/query-service/app/logparsingpipeline/collector_config.go

@@ -2,6 +2,7 @@ package logparsingpipeline
 
 import (
 	"encoding/json"
+	"slices"
 	"strings"
 	"sync"
 
@@ -24,10 +25,13 @@ var (
 )
 
 const (
-	memoryLimiterProcessor       = "memory_limiter"
-	memoryLimiterProcessorPrefix = "memory_limiter/"
-	batchProcessor               = "batch"
-	batchProcessorPrefix         = "batch/"
+	memoryLimiterProcessor        = "memory_limiter"
+	memoryLimiterProcessorPrefix  = "memory_limiter/"
+	batchProcessor                = "batch"
+	batchProcessorPrefix          = "batch/"
+	normalizerProcessor           = "signozlogsnormalizer"
+	normalizerProcessorPrefix     = "signozlogsnormalizer/"
+	normalizerBeforeUserPipelines = "before"
 )
 
 // check if the processors already exist
@@ -90,9 +94,14 @@ func getOtelPipelineFromConfig(config map[string]interface{}) (*otelPipeline, er
 //  2. signoz user-pipeline processors (in the order given by signozPipelineProcessorNames)
 //  3. custom processors (non-signoz, non-memory_limiter, non-batch processors from the current config)
 //  4. batch processors (any processor named "batch" or "batch/<id>") and anything after them
+//
+// If a signozlogsnormalizer processor is present, the order above is skipped: the first one
+// runs right after user-pipeline processors, or right before them when its config sets
+// `opamp.position: before`, and the other processors keep their places.
 func buildCollectorPipelineProcessorsList(
 	currentCollectorProcessors []string,
 	signozPipelineProcessorNames []string,
+	processorConfigs map[string]interface{},
 ) ([]string, error) {
 	lockLogsPipelineSpec.Lock()
 	defer lockLogsPipelineSpec.Unlock()
@@ -103,6 +112,23 @@ func buildCollectorPipelineProcessorsList(
 		userPipelineSet[p] = struct{}{}
 	}
 
+	if normalizerIdx := slices.IndexFunc(currentCollectorProcessors, isNormalizerProcessor); normalizerIdx >= 0 {
+		normalizerFirst := opampPosition(processorConfigs, currentCollectorProcessors[normalizerIdx]) == normalizerBeforeUserPipelines
+		result := make([]string, 0, len(currentCollectorProcessors)+len(signozPipelineProcessorNames))
+		for idx, p := range currentCollectorProcessors {
+			if idx == normalizerIdx && !normalizerFirst {
+				result = append(result, signozPipelineProcessorNames...)
+			}
+			if _, inUserPipelines := userPipelineSet[p]; !inUserPipelines && !hasSignozPipelineProcessorPrefix(p) {
+				result = append(result, p)
+			}
+			if idx == normalizerIdx && normalizerFirst {
+				result = append(result, signozPipelineProcessorNames...)
+			}
+		}
+		return result, nil
+	}
+
 	var memoryLimiters []string
 	var customProcessors []string
 	batchIdx := -1
@@ -135,13 +161,51 @@ func buildCollectorPipelineProcessorsList(
 	return result, nil
 }
 
-func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
+func isNormalizerProcessor(name string) bool {
+	return name == normalizerProcessor || strings.HasPrefix(name, normalizerProcessorPrefix)
+}
+
+// collectorNormalizesFirst reports whether the logs pipeline of the collector config runs
+// signozlogsnormalizer ahead of user pipelines, judged by the first one as in buildCollectorPipelineProcessorsList.
+// Malformed configs report false; generateCollectorConfigWithPipelines surfaces their errors.
+func collectorNormalizesFirst(collectorConf map[string]interface{}) bool {
+	p, err := getOtelPipelineFromConfig(collectorConf)
+	if err != nil || p.Pipelines.Logs == nil {
+		return false
+	}
+	normalizerIdx := slices.IndexFunc(p.Pipelines.Logs.Processors, isNormalizerProcessor)
+	if normalizerIdx < 0 {
+		return false
+	}
+	processorConfigs, _ := collectorConf["processors"].(map[string]interface{})
+	return opampPosition(processorConfigs, p.Pipelines.Logs.Processors[normalizerIdx]) == normalizerBeforeUserPipelines
+}
+
+func opampPosition(processorConfigs map[string]interface{}, name string) string {
+	config, _ := processorConfigs[name].(map[string]interface{})
+	opamp, _ := config["opamp"].(map[string]interface{})
+	position, _ := opamp["position"].(string)
+	return position
+}
+
+func unmarshalCollectorConfig(config []byte) (map[string]interface{}, error) {
 	var collectorConf map[string]interface{}
-	err := yaml.Unmarshal([]byte(config), &collectorConf)
-	if err != nil {
+	if err := yaml.Unmarshal(config, &collectorConf); err != nil {
 		return nil, errors.WrapInvalidInputf(err, CodeCollectorConfigUnmarshalFailed, "could not unmarshal collector config")
 	}
+	return collectorConf, nil
+}
 
+func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
+	collectorConf, err := unmarshalCollectorConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	return generateCollectorConfigWithPipelines(collectorConf, pipelines)
+}
+
+// generateCollectorConfigWithPipelines modifies collectorConf in place.
+func generateCollectorConfigWithPipelines(collectorConf map[string]any, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
 	signozPipelineProcessors, signozPipelineProcNames, err := PreparePipelineProcessor(pipelines)
 	if err != nil {
 		return nil, err
@@ -180,11 +244,11 @@ func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetyp
 		return nil, errors.NewInternalf(CodeCollectorConfigLogsPipelineNotFound, "logs pipeline doesn't exist")
 	}
 
-	updatedProcessorList, _ := buildCollectorPipelineProcessorsList(p.Pipelines.Logs.Processors, signozPipelineProcNames)
+	updatedProcessorList, _ := buildCollectorPipelineProcessorsList(p.Pipelines.Logs.Processors, signozPipelineProcNames, collectorConf["processors"].(map[string]any))
 	p.Pipelines.Logs.Processors = updatedProcessorList
 
 	// add the new processor to the data ( no checks required as the keys will exists)
-	collectorConf["service"].(map[string]interface{})["pipelines"].(map[string]interface{})["logs"] = p.Pipelines.Logs
+	collectorConf["service"].(map[string]any)["pipelines"].(map[string]any)["logs"] = p.Pipelines.Logs
 
 	updatedConf, err := yaml.Marshal(collectorConf)
 	if err != nil {

pkg/query-service/app/logparsingpipeline/collector_config_test.go

@@ -5,11 +5,14 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	"github.com/SigNoz/signoz/pkg/query-service/model"
 	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
 	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
 	. "github.com/smartystreets/goconvey/convey"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 )
@@ -206,13 +209,74 @@ var BuildLogsPipelineTestData = []struct {
 func TestBuildLogsPipeline(t *testing.T) {
 	for _, test := range BuildLogsPipelineTestData {
 		Convey(test.Name, t, func() {
-			v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines)
+			v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines, nil)
 			So(err, ShouldBeNil)
 			So(v, ShouldResemble, test.finalOutput)
 		})
 	}
 }
 
+func TestBuildLogsPipelineWithNormalizer(t *testing.T) {
+	a := constants.LogsPPLPfx + "a"
+	b := constants.LogsPPLPfx + "b"
+	before := map[string]any{"opamp": map[string]any{"position": "before"}}
+	after := map[string]any{"opamp": map[string]any{"position": "after"}}
+
+	testCases := []struct {
+		name          string
+		fromCollector []string
+		userPipelines []string
+		configs       map[string]any
+		expected      []string
+	}{
+		{
+			name:          "ExistingAndStaleUserProcessorsReplaced",
+			fromCollector: []string{"memory_limiter", a, constants.LogsPPLPfx + "stale", constants.OldLogsPPLPfx + "old", "signozlogsnormalizer", "custom", "batch"},
+			userPipelines: []string{a, b},
+			expected:      []string{"memory_limiter", a, b, "signozlogsnormalizer", "custom", "batch"},
+		},
+		{
+			name:          "NamedNormalizerAfter_CustomProcessorsKeepPlace",
+			fromCollector: []string{"memory_limiter", "custom_a", "signozlogsnormalizer/x", a, "custom_b", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/x": after},
+			expected:      []string{"memory_limiter", "custom_a", a, "signozlogsnormalizer/x", "custom_b", "batch"},
+		},
+		{
+			name:          "FirstNormalizerAfter_SecondIgnored",
+			fromCollector: []string{"memory_limiter", "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/first": after, "signozlogsnormalizer/second": before},
+			expected:      []string{"memory_limiter", a, "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+		},
+		{
+			name:          "FirstNormalizerBefore_SecondIgnored",
+			fromCollector: []string{"memory_limiter", "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/first": before, "signozlogsnormalizer/second": after},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer/first", a, "signozlogsnormalizer/second", "batch"},
+		},
+		{
+			name:          "NormalizerAfterBatch",
+			fromCollector: []string{"memory_limiter", "batch", "signozlogsnormalizer"},
+			userPipelines: []string{a},
+			expected:      []string{"memory_limiter", "batch", a, "signozlogsnormalizer"},
+		},
+		{
+			name:          "NoUserPipelines_StaleRemoved",
+			fromCollector: []string{"memory_limiter", a, "signozlogsnormalizer", "batch"},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer", "batch"},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			processors, err := buildCollectorPipelineProcessorsList(testCase.fromCollector, testCase.userPipelines, testCase.configs)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected, processors)
+		})
+	}
+}
+
 func TestPipelineAliasCollisionsDontResultInDuplicateCollectorProcessors(t *testing.T) {
 	require := require.New(t)
 
@@ -437,3 +501,61 @@ func TestPipeCharInAliasDoesntBreakCollectorConfig(t *testing.T) {
 		}, result[0].Attributes_string,
 	)
 }
+
+func TestRecommendAgentConfigWithNormalizerProcessorAndJSONBody(t *testing.T) {
+	userPipeline := makeTestAddAttributePipeline()
+	controller := &LogParsingPipelineController{
+		fl: flaggertest.WithUseJSONBody(t, true),
+		GetIntegrationPipelines: func(context.Context, string) ([]pipelinetypes.GettablePipeline, error) {
+			return []pipelinetypes.GettablePipeline{userPipeline}, nil
+		},
+	}
+	normalizeOperator := CollectorConfProcessorName(getNormalizePipeline())
+	user := CollectorConfProcessorName(userPipeline)
+
+	testCases := []struct {
+		name       string
+		processors string
+		logs       string
+		expected   []any
+	}{
+		{
+			name:     "NoNormalizer_OperatorPrepended",
+			logs:     "[memory_limiter, batch]",
+			expected: []any{"memory_limiter", normalizeOperator, user, "batch"},
+		},
+		{
+			name:     "NormalizerDefault_OperatorKept",
+			logs:     "[memory_limiter, signozlogsnormalizer, batch]",
+			expected: []any{"memory_limiter", normalizeOperator, user, "signozlogsnormalizer", "batch"},
+		},
+		{
+			name:       "NormalizerBefore_OperatorDropped",
+			processors: "signozlogsnormalizer:\n    opamp:\n      position: before\n",
+			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
+			expected:   []any{"memory_limiter", "signozlogsnormalizer", user, "batch"},
+		},
+		{
+			name:       "NormalizerAfter_OperatorKept",
+			processors: "signozlogsnormalizer:\n    opamp:\n      position: after\n",
+			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
+			expected:   []any{"memory_limiter", normalizeOperator, user, "signozlogsnormalizer", "batch"},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			conf := fmt.Sprintf(
+				"processors:\n  memory_limiter: {}\n  batch: {}\n  %s\nservice:\n  pipelines:\n    logs:\n      receivers: [otlp]\n      processors: %s\n      exporters: [clickhouselogsexporter]\n",
+				testCase.processors, testCase.logs,
+			)
+
+			recommended, _, err := controller.RecommendAgentConfig(valuer.GenerateUUID(), []byte(conf), nil)
+			require.NoError(t, err)
+
+			var recommendedConf map[string]any
+			require.NoError(t, yaml.Unmarshal(recommended, &recommendedConf))
+			logs := recommendedConf["service"].(map[string]any)["pipelines"].(map[string]any)["logs"].(map[string]any)
+			assert.Equal(t, testCase.expected, logs["processors"])
+		})
+	}
+}

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

@@ -220,26 +220,17 @@ func (ic *LogParsingPipelineController) ValidatePipelines(ctx context.Context,
 	return err
 }
 
-// withNormalizePipeline places normalize where the read path dictates. Ahead of user pipelines
-// when queries run on body_v2 (use_json_body), so operators see the body the explorer shows.
-// After them when dual ingestion alone writes body_v2, so operators keep seeing the raw body
-// users still query. Absent when neither flag is on.
-func (ic *LogParsingPipelineController) withNormalizePipeline(ctx context.Context, orgID valuer.UUID, pipelines []pipelinetypes.GettablePipeline) []pipelinetypes.GettablePipeline {
-	evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
-	dualIngestion := ic.fl.BooleanOrEmpty(ctx, flagger.FeatureJSONBodyDualIngestion, evalCtx)
-	switch {
-	case ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, evalCtx):
-		return append([]pipelinetypes.GettablePipeline{getNormalizePipeline(dualIngestion)}, pipelines...)
-	case dualIngestion:
-		return append(slices.Clone(pipelines), getNormalizePipeline(true))
-	default:
+// withNormalizePipeline places normalize ahead of user pipelines when queries run on body_v2
+// (use_json_body), so operators see the body the explorer shows. Absent when the flag is off
+// or collectorNormalizesFirst.
+func (ic *LogParsingPipelineController) withNormalizePipeline(ctx context.Context, orgID valuer.UUID, pipelines []pipelinetypes.GettablePipeline, collectorNormalizesFirst bool) []pipelinetypes.GettablePipeline {
+	if collectorNormalizesFirst || !ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
 		return pipelines
 	}
+	return append([]pipelinetypes.GettablePipeline{getNormalizePipeline()}, pipelines...)
 }
 
-// stashOriginalBody makes normalize carry the pre-normalization body in an internal attribute
-// for the exporter to restore into the legacy body column.
-func getNormalizePipeline(stashOriginalBody bool) pipelinetypes.GettablePipeline {
+func getNormalizePipeline() pipelinetypes.GettablePipeline {
 	return pipelinetypes.GettablePipeline{
 		StoreablePipeline: pipelinetypes.StoreablePipeline{
 			Name:    "Default Pipeline - PreProcessing Body",
@@ -258,11 +249,10 @@ func getNormalizePipeline(stashOriginalBody bool) pipelinetypes.GettablePipeline
 		},
 		Config: []pipelinetypes.PipelineOperator{
 			{
-				ID:                    uuid.NewString(),
-				Type:                  "normalize",
-				Enabled:               true,
-				If:                    "body != nil",
-				JSONBodyDualIngestion: stashOriginalBody,
+				ID:      uuid.NewString(),
+				Type:    "normalize",
+				Enabled: true,
+				If:      "body != nil",
 			},
 		},
 	}
@@ -371,11 +361,8 @@ func (ic *LogParsingPipelineController) PreviewLogsPipelines(
 	}
 
 	// The collector gets the same pipeline prepended over opamp; see RecommendAgentConfig.
-	// Under dual ingestion alone it runs after user operators and only feeds body_v2, which
-	// the explorer does not show yet, so the preview leaves it out. The original-body stash
-	// is left off: the preview has no exporter to restore and strip it.
 	if ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-		pipelines = append([]pipelinetypes.GettablePipeline{getNormalizePipeline(false)}, pipelines...)
+		pipelines = append([]pipelinetypes.GettablePipeline{getNormalizePipeline()}, pipelines...)
 	}
 
 	result, collectorLogs, err := SimulatePipelinesProcessing(ctx, pipelines, request.Logs)
@@ -396,14 +383,14 @@ func (pc *LogParsingPipelineController) AgentFeatureType() agentConf.AgentFeatur
 
 // Implements agentConf.AgentFeature interface.
 // RecommendAgentConfig generates the collector config to be sent to agents.
-// The normalize pipeline (when use_json_body or json_body_dual_ingestion is on) is placed
+// The normalize pipeline (when use_json_body is on) is placed
 // here, after rawPipelineData is serialized. So it is only present in the config sent to
 // the collector and never persisted to the database as part of the user's pipeline list.
 //
 // NOTE: The configId sent to agents is derived from the pipeline version number
 // (e.g. "LogPipelines:5"), not the YAML content. If server-side logic changes
-// the generated YAML without bumping the version (e.g. toggling the use_json_body or
-// json_body_dual_ingestion flags or updating operator IfExpressions), agents that already
+// the generated YAML without bumping the version (e.g. toggling the use_json_body
+// flag or updating operator IfExpressions), agents that already
 // applied that version will not re-apply the new config. In such cases, users must save a
 // new pipeline version via the API to force agents to pick up the change.
 func (pc *LogParsingPipelineController) RecommendAgentConfig(
@@ -431,10 +418,15 @@ func (pc *LogParsingPipelineController) RecommendAgentConfig(
 		return nil, "", err
 	}
 
-	// normalize is only for sending to the collector, never persisted
-	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines)
+	collectorConf, err := unmarshalCollectorConfig(currentConfYaml)
+	if err != nil {
+		return nil, "", err
+	}
 
-	updatedConf, err := GenerateCollectorConfigWithPipelines(currentConfYaml, enrichedPipelines)
+	// normalize is only for sending to the collector, never persisted
+	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines, collectorNormalizesFirst(collectorConf))
+
+	updatedConf, err := generateCollectorConfigWithPipelines(collectorConf, enrichedPipelines)
 	if err != nil {
 		return nil, "", err
 	}

pkg/signoz/provider.go

@@ -262,6 +262,7 @@ func NewSQLMigrationProviderFactories(
 		sqlmigration.NewAddRuleViewFactory(sqlstore, sqlschema),
 		sqlmigration.NewAddFieldTuplesFactory(sqlstore),
 		sqlmigration.NewKeepRelevantLLMPricingRulesFactory(),
+		sqlmigration.NewAddGCPConnectionMethodFactory(),
 	)
 }
 

pkg/sqlmigration/136_add_gcp_connection_method.go

@@ -0,0 +1,85 @@
+package sqlmigration
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/uptrace/bun"
+	"github.com/uptrace/bun/migrate"
+)
+
+type addGCPConnectionMethod struct {
+	logger *slog.Logger
+}
+
+type gcpConnectionMethodMigrationRow struct {
+	bun.BaseModel `bun:"table:cloud_integration"`
+
+	ID     string `bun:"id"`
+	Config string `bun:"config"`
+}
+
+func NewAddGCPConnectionMethodFactory() factory.ProviderFactory[SQLMigration, Config] {
+	return factory.NewProviderFactory(factory.MustNewName("add_gcp_connection_method"), func(ctx context.Context, ps factory.ProviderSettings, c Config) (SQLMigration, error) {
+		return &addGCPConnectionMethod{logger: ps.Logger}, nil
+	})
+}
+
+func (migration *addGCPConnectionMethod) Register(migrations *migrate.Migrations) error {
+	return migrations.Register(migration.Up, migration.Down)
+}
+
+func (migration *addGCPConnectionMethod) Up(ctx context.Context, db *bun.DB) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	var accounts []*gcpConnectionMethodMigrationRow
+	if err := tx.NewSelect().
+		Model(&accounts).
+		Where("provider = ?", "gcp").
+		Scan(ctx); err != nil {
+		return err
+	}
+
+	for _, account := range accounts {
+		// Preserve config fields that are not known to this migration.
+		var config map[string]json.RawMessage
+		if err := json.Unmarshal([]byte(account.Config), &config); err != nil {
+			migration.logger.WarnContext(ctx, "skipping GCP account with unreadable config", slog.String("account_id", account.ID), errors.Attr(err))
+			continue
+		}
+		if config == nil {
+			migration.logger.WarnContext(ctx, "skipping GCP account with null config", slog.String("account_id", account.ID))
+			continue
+		}
+		if _, exists := config["connectionMethod"]; exists {
+			continue
+		}
+
+		config["connectionMethod"] = json.RawMessage(`"manual"`)
+		encoded, err := json.Marshal(config)
+		if err != nil {
+			return err
+		}
+
+		if _, err := tx.NewUpdate().
+			Model((*gcpConnectionMethodMigrationRow)(nil)).
+			Set("config = ?", string(encoded)).
+			Where("id = ?", account.ID).
+			Exec(ctx); err != nil {
+			return err
+		}
+	}
+
+	return tx.Commit()
+}
+
+func (migration *addGCPConnectionMethod) Down(context.Context, *bun.DB) error {
+	return nil
+}

pkg/types/cloudintegrationtypes/account.go

@@ -150,7 +150,7 @@ func NewAccountFromStorable(storableAccount *StorableCloudIntegration) (*Account
 		}
 		account.Config.Azure = azureConfig
 	case CloudProviderTypeGCP:
-		gcpConfig := new(GCPAccountConfig)
+		gcpConfig := &GCPAccountConfig{ConnectionMethod: GCPConnectionMethodManual}
 		err := json.Unmarshal([]byte(storableAccount.Config), gcpConfig)
 		if err != nil {
 			return nil, err
@@ -240,25 +240,16 @@ func NewAccountConfigFromPostable(provider CloudProviderType, config *PostableAc
 			return nil, errors.NewInvalidInputf(ErrCodeInvalidInput, "GCP config can not be nil for GCP provider")
 		}
 
-		if config.GCP.DeploymentProjectID == "" {
-			return nil, errors.NewInvalidInputf(ErrCodeInvalidInput, "deployment project ID is required for GCP provider")
+		gcpConfig := &GCPAccountConfig{
+			ConnectionMethod:    config.GCP.ConnectionMethod,
+			DeploymentProjectID: config.GCP.DeploymentProjectID,
+			ProjectIDs:          config.GCP.ProjectIDs,
+			DeploymentRegion:    config.GCP.DeploymentRegion,
 		}
-
-		if err := validateGCPRegion(config.GCP.DeploymentRegion); err != nil {
+		if err := gcpConfig.ErrIfInvalid(); err != nil {
 			return nil, err
 		}
-
-		if len(config.GCP.ProjectIDs) == 0 {
-			return nil, errors.NewInvalidInputf(ErrCodeInvalidInput, "at least one project id is required for GCP provider")
-		}
-
-		return &AccountConfig{
-			GCP: &GCPAccountConfig{
-				DeploymentProjectID: config.GCP.DeploymentProjectID,
-				ProjectIDs:          config.GCP.ProjectIDs,
-				DeploymentRegion:    config.GCP.DeploymentRegion,
-			},
-		}, nil
+		return &AccountConfig{GCP: gcpConfig}, nil
 	default:
 		return nil, errors.NewInvalidInputf(ErrCodeCloudProviderInvalidInput, "invalid cloud provider: %s", provider.StringValue())
 	}
@@ -301,12 +292,12 @@ func NewAccountConfigFromUpdatable(provider CloudProviderType, config *Updatable
 			return nil, err
 		}
 
-		if len(config.Config.GCP.ProjectIDs) == 0 {
-			return nil, errors.NewInvalidInputf(ErrCodeInvalidInput, "at least one project id is required for GCP provider")
+		if err := validateGCPProjectIDs(config.Config.GCP.ProjectIDs); err != nil {
+			return nil, err
 		}
 
-		if config.Config.GCP.DeploymentProjectID == "" {
-			return nil, errors.NewInvalidInputf(ErrCodeInvalidInput, "deployment project ID is required for GCP provider")
+		if err := validateGCPDeploymentProjectID(config.Config.GCP.DeploymentProjectID); err != nil {
+			return nil, err
 		}
 
 		return &AccountConfig{
@@ -355,6 +346,13 @@ func GetSigNozAPIURLFromDeployment(deployment *zeustypes.GettableDeployment) (st
 }
 
 func (account *Account) Update(provider CloudProviderType, config *AccountConfig) error {
+	if provider == CloudProviderTypeGCP {
+		if account.Config.GCP.ConnectionMethod == GCPConnectionMethodAgent {
+			return errors.NewInvalidInputf(ErrCodeInvalidInput, "GCP agent account configuration cannot be updated")
+		}
+		config.GCP.ConnectionMethod = account.Config.GCP.ConnectionMethod
+	}
+
 	// deployment region can not be updated once set for Azure
 	if provider == CloudProviderTypeAzure {
 		config.Azure.DeploymentRegion = account.Config.Azure.DeploymentRegion
@@ -366,6 +364,18 @@ func (account *Account) Update(provider CloudProviderType, config *AccountConfig
 	return nil
 }
 
+func (account *Account) ErrIfInvalidProviderAccountID(providerAccountID string) error {
+	if account.Provider != CloudProviderTypeGCP || account.Config.GCP.ConnectionMethod != GCPConnectionMethodAgent {
+		return nil
+	}
+
+	if providerAccountID != account.Config.GCP.DeploymentProjectID {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "providerAccountId must match deploymentProjectId for GCP agent setup")
+	}
+
+	return nil
+}
+
 func (account *Account) UpdateAgentReport(providerAccountID *string, agentReport *AgentReport) {
 	account.ProviderAccountID = providerAccountID
 	account.AgentReport = agentReport

pkg/types/cloudintegrationtypes/cloudprovider_gcp.go

@@ -1,30 +1,95 @@
 package cloudintegrationtypes
 
+import (
+	"bytes"
+	"fmt"
+	"slices"
+	"strings"
+	"text/template"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+var (
+	GCPConnectionMethodManual = GCPConnectionMethod{valuer.NewString("manual")}
+	GCPConnectionMethodAgent  = GCPConnectionMethod{valuer.NewString("agent")}
+
+	// TODO: Confirm the published bootstrap module location before release.
+	AgentGCPTerraformModuleStorePath = "gs://signoz-cloud-integration/gcp/%s/bootstrap/"
+	AgentGCPBootstrapDeploymentName  = "signoz-integration-bootstrap"
+	gcpBootstrapExecutorName         = "signoz-integration-bootstrap"
+	gcpAPIKeySecretID                = "signoz-integration-credentials-api-key"
+	gcpIngestionKeySecretID          = "signoz-integration-credentials-ingestion-key"
+
+	gcpCLITemplate = template.Must(template.New("gcpCLI").Funcs(template.FuncMap{
+		"shellQuote": func(value string) string {
+			return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
+		},
+	}).Parse(gcpCLITemplateStr()))
+	gcpCreateServiceAccountTemplate = template.Must(gcpCLITemplate.New("gcpCreateServiceAccount").Parse(gcpCreateServiceAccountTemplateStr()))
+	gcpCreateSecretsTemplate        = template.Must(gcpCLITemplate.New("gcpCreateSecrets").Parse(gcpCreateSecretsTemplateStr()))
+)
+
+type GCPConnectionMethod struct{ valuer.String }
+
+func (GCPConnectionMethod) Enum() []any {
+	return []any{GCPConnectionMethodManual, GCPConnectionMethodAgent}
+}
+
 type GCPAccountConfig struct {
-	// Project ID where central pub/sub for logs exist
+	ConnectionMethod GCPConnectionMethod `json:"connectionMethod" required:"true"`
+	// DeploymentProjectID is the project hosting the integration resources.
 	DeploymentProjectID string `json:"deploymentProjectId" required:"true"`
-	// Project ID where otel collector will be deployed
+	// DeploymentRegion is the region used for the integration resources.
 	DeploymentRegion string `json:"deploymentRegion" required:"true"`
-	// List of project IDs to monitor
+	// Agent accounts store exactly the deployment project; manual accounts may monitor multiple projects.
 	ProjectIDs []string `json:"projectIds" required:"true" nullable:"false"`
 }
 
-type GCPPostableAccountConfig = GCPAccountConfig
-
-type UpdatableGCPAccountConfig struct {
-	// Project ID where central pub/sub for logs exist
+type GCPPostableAccountConfig struct {
+	ConnectionMethod GCPConnectionMethod `json:"connectionMethod" required:"true"`
+	// DeploymentProjectID is the project hosting the integration resources.
 	DeploymentProjectID string `json:"deploymentProjectId" required:"true"`
-	// Compute service region where otel collector will be deployed
+	// DeploymentRegion is the region used for the integration resources.
 	DeploymentRegion string `json:"deploymentRegion" required:"true"`
-	// List of project IDs to monitor
-	ProjectIDs []string `json:"projectIds" required:"true"`
+	// Agent setup requires exactly DeploymentProjectID; manual setup may select multiple projects.
+	ProjectIDs []string `json:"projectIds" required:"true" nullable:"false"`
 }
 
-type GCPConnectionArtifact struct{}
+// UpdatableGCPAccountConfig contains settings for manual integrations only.
+type UpdatableGCPAccountConfig struct {
+	// DeploymentProjectID is the project hosting the manually configured integration resources.
+	DeploymentProjectID string `json:"deploymentProjectId" required:"true"`
+	// DeploymentRegion is the region used for the manually configured integration resources.
+	DeploymentRegion string `json:"deploymentRegion" required:"true"`
+	// ProjectIDs lists the projects monitored by the manual integration.
+	ProjectIDs []string `json:"projectIds" required:"true" nullable:"false"`
+}
 
-type GCPIntegrationConfig struct{}
+type GCPConnectionArtifact struct {
+	CLICommand                  string `json:"cliCommand" required:"true"`
+	CreateSecretsCommand        string `json:"createSecretsCommand" required:"true"`
+	CreateServiceAccountCommand string `json:"createServiceAccountCommand" required:"true"`
+}
 
-type GCPTelemetryCollectionStrategy struct{}
+type GCPIntegrationConfig struct {
+	GCPAccountConfig
+	TelemetryCollectionStrategy *GCPTelemetryCollectionStrategy `json:"telemetryCollectionStrategy" required:"true" nullable:"false"`
+}
+
+type GCPTelemetryCollectionStrategy struct {
+	Metrics *GCPMetricsCollectionStrategy `json:"metrics,omitempty" required:"false" nullable:"false"`
+	Logs    *GCPLogsCollectionStrategy    `json:"logs,omitempty" required:"false" nullable:"false"`
+}
+
+type GCPMetricsCollectionStrategy struct {
+	MetricNames []string `json:"metricNames" required:"true" nullable:"false"`
+}
+
+type GCPLogsCollectionStrategy struct {
+	SinkFilter string `json:"sinkFilter" required:"true"`
+}
 
 type GCPServiceConfig struct {
 	Logs    *GCPServiceLogsConfig    `json:"logs,omitempty" required:"false"`
@@ -38,3 +103,219 @@ type GCPServiceLogsConfig struct {
 type GCPServiceMetricsConfig struct {
 	Enabled bool `json:"enabled" required:"true"`
 }
+
+func (config *GCPAccountConfig) ErrIfInvalid() error {
+	if config.ConnectionMethod.IsZero() {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "connectionMethod is required for GCP provider")
+	}
+	if config.ConnectionMethod != GCPConnectionMethodManual && config.ConnectionMethod != GCPConnectionMethodAgent {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "unsupported GCP connectionMethod: %q", config.ConnectionMethod.StringValue())
+	}
+	if err := validateGCPDeploymentProjectID(config.DeploymentProjectID); err != nil {
+		return err
+	}
+	if err := validateGCPRegion(config.DeploymentRegion); err != nil {
+		return err
+	}
+	if err := validateGCPProjectIDs(config.ProjectIDs); err != nil {
+		return err
+	}
+	if config.ConnectionMethod == GCPConnectionMethodAgent && (len(config.ProjectIDs) != 1 || config.ProjectIDs[0] != config.DeploymentProjectID) {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "projectIds must contain only deploymentProjectId for GCP agent setup")
+	}
+	return nil
+}
+
+func NewGCPConnectionArtifact(accountID valuer.UUID, agentVersion string, creds *Credentials, cfg *GCPAccountConfig) (*GCPConnectionArtifact, error) {
+	executorEmail := fmt.Sprintf("%s@%s.iam.gserviceaccount.com", gcpBootstrapExecutorName, cfg.DeploymentProjectID)
+	data := gcpTemplateData{
+		DeploymentName:    fmt.Sprintf("projects/%s/locations/%s/deployments/%s", cfg.DeploymentProjectID, cfg.DeploymentRegion, AgentGCPBootstrapDeploymentName),
+		ProjectID:         cfg.DeploymentProjectID,
+		Region:            cfg.DeploymentRegion,
+		ServiceAccount:    fmt.Sprintf("projects/%s/serviceAccounts/%s", cfg.DeploymentProjectID, executorEmail),
+		ExecutorName:      gcpBootstrapExecutorName,
+		ExecutorEmail:     executorEmail,
+		APIKeySecretID:    gcpAPIKeySecretID,
+		IngestionSecretID: gcpIngestionKeySecretID,
+		APIKey:            creds.SigNozAPIKey,
+		IngestionKey:      creds.IngestionKey,
+		ModuleURL:         fmt.Sprintf(AgentGCPTerraformModuleStorePath, agentVersion),
+		InputValues: fmt.Sprintf(
+			"project_id=%s,deployment_region=%s,integration_account_id=%s,signoz_api_url=%s,ingestion_url=%s,"+
+				"api_key_secret_id=%s,ingestion_key_secret_id=%s",
+			cfg.DeploymentProjectID, cfg.DeploymentRegion, accountID.StringValue(), creds.SigNozAPIURL, creds.IngestionURL, gcpAPIKeySecretID, gcpIngestionKeySecretID,
+		),
+	}
+
+	cliCommand, err := newGCPConnectionCommand(gcpCLITemplate, data)
+	if err != nil {
+		return nil, err
+	}
+	createServiceAccountCommand, err := newGCPConnectionCommand(gcpCreateServiceAccountTemplate, data)
+	if err != nil {
+		return nil, err
+	}
+	createSecretsCommand, err := newGCPConnectionCommand(gcpCreateSecretsTemplate, data)
+	if err != nil {
+		return nil, err
+	}
+
+	return &GCPConnectionArtifact{
+		CLICommand:                  cliCommand,
+		CreateServiceAccountCommand: createServiceAccountCommand,
+		CreateSecretsCommand:        createSecretsCommand,
+	}, nil
+}
+
+// NewGCPIntegrationConfig sorts and deduplicates the input slices in place and formats sink filters for joining with OR.
+func NewGCPIntegrationConfig(config *GCPAccountConfig, metricNames, sinkFilters []string) *GCPIntegrationConfig {
+	// Parentheses preserve each service filter's meaning when combined with OR.
+	for i, filter := range sinkFilters {
+		sinkFilters[i] = "(" + strings.TrimSpace(filter) + ")"
+	}
+
+	// Keep the generated config stable and avoid collecting shared metrics more than once.
+	slices.Sort(metricNames)
+	metricNames = slices.Compact(metricNames)
+	slices.Sort(sinkFilters)
+	sinkFilters = slices.Compact(sinkFilters)
+
+	strategy := new(GCPTelemetryCollectionStrategy)
+	if len(metricNames) > 0 {
+		strategy.Metrics = &GCPMetricsCollectionStrategy{MetricNames: metricNames}
+	}
+	if len(sinkFilters) > 0 {
+		strategy.Logs = &GCPLogsCollectionStrategy{SinkFilter: strings.Join(sinkFilters, " OR ")}
+	}
+
+	return &GCPIntegrationConfig{
+		GCPAccountConfig:            *config,
+		TelemetryCollectionStrategy: strategy,
+	}
+}
+
+type gcpTemplateData struct {
+	DeploymentName    string
+	ProjectID         string
+	Region            string
+	ServiceAccount    string
+	ExecutorName      string
+	ExecutorEmail     string
+	APIKeySecretID    string
+	IngestionSecretID string
+	APIKey            string
+	IngestionKey      string
+	ModuleURL         string
+	InputValues       string
+}
+
+func newGCPConnectionCommand(commandTemplate *template.Template, data gcpTemplateData) (string, error) {
+	var buf bytes.Buffer
+	if err := commandTemplate.Execute(&buf, data); err != nil {
+		return "", err
+	}
+	return buf.String(), nil
+}
+
+func gcpCLITemplateStr() string {
+	return `gcloud infra-manager deployments apply \
+  {{.DeploymentName | shellQuote}} \
+  --project={{.ProjectID | shellQuote}} \
+  --service-account={{.ServiceAccount | shellQuote}} \
+  --gcs-source={{.ModuleURL | shellQuote}} \
+  --tf-version-constraint='=1.5.7' \
+  --input-values={{.InputValues | shellQuote}},api_key_secret_version="${api_key_version:-latest}",ingestion_key_secret_version="${ingestion_key_version:-latest}"`
+}
+
+func gcpCreateServiceAccountTemplateStr() string {
+	return `project_id={{.ProjectID | shellQuote}}
+executor_email={{.ExecutorEmail | shellQuote}}
+api_secret_id={{.APIKeySecretID | shellQuote}}
+ingestion_secret_id={{.IngestionSecretID | shellQuote}}
+caller_email=$(gcloud config get-value account)
+
+gcloud iam service-accounts create {{.ExecutorName | shellQuote}} \
+  --project="$project_id" --display-name="SigNoz bootstrap executor"
+
+for executor_role in \
+  roles/config.agent \
+  roles/serviceusage.serviceUsageAdmin \
+  roles/resourcemanager.projectIamAdmin \
+  roles/iam.serviceAccountAdmin \
+  roles/iam.serviceAccountUser \
+  roles/run.admin \
+  roles/pubsub.admin \
+  roles/logging.configWriter
+do
+  gcloud projects add-iam-policy-binding "$project_id" \
+    --member="serviceAccount:${executor_email}" --role="$executor_role" \
+    --condition=None
+done
+
+for secret_id in "$api_secret_id" "$ingestion_secret_id"
+do
+  gcloud secrets add-iam-policy-binding "$secret_id" \
+    --project="$project_id" --member="serviceAccount:${executor_email}" \
+    --role=roles/secretmanager.admin --condition=None
+done
+
+gcloud iam service-accounts add-iam-policy-binding "$executor_email" \
+  --project="$project_id" --member="user:${caller_email}" \
+  --role=roles/iam.serviceAccountUser --condition=None`
+}
+
+func gcpCreateSecretsTemplateStr() string {
+	return `project_id={{.ProjectID | shellQuote}}
+region={{.Region | shellQuote}}
+api_secret_id={{.APIKeySecretID | shellQuote}}
+ingestion_secret_id={{.IngestionSecretID | shellQuote}}
+
+gcloud services enable config.googleapis.com cloudbuild.googleapis.com \
+  storage.googleapis.com iam.googleapis.com serviceusage.googleapis.com \
+  cloudresourcemanager.googleapis.com secretmanager.googleapis.com \
+  --project="$project_id" &&
+sleep 60 &&
+
+gcloud secrets create "$api_secret_id" \
+  --project="$project_id" --replication-policy=user-managed \
+  --locations="$region" --labels=managed-by=signoz-integration &&
+
+gcloud secrets create "$ingestion_secret_id" \
+  --project="$project_id" --replication-policy=user-managed \
+  --locations="$region" --labels=managed-by=signoz-integration &&
+
+api_key_version=$(printf '%s' {{.APIKey | shellQuote}} | \
+  gcloud secrets versions add "$api_secret_id" \
+    --project="$project_id" --data-file=- --format='value(name.basename())') &&
+
+ingestion_key_version=$(printf '%s' {{.IngestionKey | shellQuote}} | \
+  gcloud secrets versions add "$ingestion_secret_id" \
+    --project="$project_id" --data-file=- --format='value(name.basename())')`
+}
+
+func validateGCPDeploymentProjectID(projectID string) error {
+	if strings.TrimSpace(projectID) == "" {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "deploymentProjectId is required for GCP provider")
+	}
+	if projectID != strings.TrimSpace(projectID) {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "deploymentProjectId cannot contain leading or trailing whitespace")
+	}
+	return nil
+}
+
+func validateGCPProjectIDs(projectIDs []string) error {
+	if len(projectIDs) == 0 {
+		return errors.NewInvalidInputf(ErrCodeInvalidInput, "projectIds must contain at least one project ID for GCP provider")
+	}
+
+	for _, projectID := range projectIDs {
+		if strings.TrimSpace(projectID) == "" {
+			return errors.NewInvalidInputf(ErrCodeInvalidInput, "projectIds cannot contain an empty or whitespace-only project ID")
+		}
+		if projectID != strings.TrimSpace(projectID) {
+			return errors.NewInvalidInputf(ErrCodeInvalidInput, "projectIds cannot contain leading or trailing whitespace")
+		}
+	}
+
+	return nil
+}

pkg/types/pipelinetypes/pipeline.go

@@ -122,9 +122,6 @@ type PipelineOperator struct {
 	EnablePaths        bool   `json:"enable_paths,omitempty" yaml:"enable_paths,omitempty"`
 	PathPrefix         string `json:"path_prefix,omitempty" yaml:"path_prefix,omitempty"`
 
-	// normalize fields, set by the server only
-	JSONBodyDualIngestion bool `json:"-" yaml:"json_body_dual_ingestion,omitempty"`
-
 	// Used in Severity Parsing and JSON Flattening mapping
 	Mapping map[string][]string `json:"mapping,omitempty" yaml:"mapping,omitempty"`
 	// severity parser fields

tests/fixtures/cloudintegrations.py

@@ -49,8 +49,7 @@ class ProviderAccountSpec:
 class ProviderServiceSpec:
     provider: str
     service_id: str
-    # GCP ships every service with supportedSignals.logs false, so a logs block
-    # is neither required on write nor persisted.
+    # Whether writes require a logs block and reads include it.
     supports_logs: bool
     account_config: dict
     # id shown in parametrized test names; defaults to the provider slug.

tests/integration/tests/alertmanager/09_googlechat.py

@@ -22,7 +22,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/11_incidentio.py

@@ -23,7 +23,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/13_jira.py

@@ -24,7 +24,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/14_jira_test_channel.py

@@ -30,7 +30,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 # channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
 # with a hardcoded firing test alert and no retry. The search stub decides which
 # branch runs (create / update / reopen), so the whole issue lifecycle is

tests/integration/tests/alertmanager/15_jsmops.py

@@ -25,7 +25,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/16_jsmops_test_channel.py

@@ -24,7 +24,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 # channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
 # with a hardcoded firing test alert and no retry: create alert on the JSM Ops
 # gateway, then append a timeline note. Default-template events + retry are in

tests/integration/tests/cloudintegrations/02_create_account.py

@@ -1,3 +1,5 @@
+import os
+import subprocess
 from collections.abc import Callable
 from http import HTTPStatus
 
@@ -6,7 +8,7 @@ import requests
 
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD, add_license
-from fixtures.cloudintegrations import ProviderAccountSpec
+from fixtures.cloudintegrations import ProviderAccountSpec, simulate_agent_checkin
 from fixtures.logger import setup_logger
 
 logger = setup_logger(__name__)
@@ -27,12 +29,14 @@ GCP_ACCOUNT_SPEC = ProviderAccountSpec(
     },
     build_config=lambda p: {
         "gcp": {
+            "connectionMethod": "manual",
             "deploymentProjectId": p["deployment_project_id"],
             "deploymentRegion": p["deployment_region"],
             "projectIds": p["project_ids"],
         }
     },
     expected_config=lambda p: {
+        "connectionMethod": "manual",
         "deploymentProjectId": p["deployment_project_id"],
         "deploymentRegion": p["deployment_region"],
         "projectIds": p["project_ids"],
@@ -91,6 +95,167 @@ def test_create_account(
         assert artifact.get("gcp") is None, f"GCP should not return a connection artifact, got: {artifact}"
 
 
+def test_create_gcp_manual_account_connection_method(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_cloud_integration_account: Callable,
+) -> None:
+    """Manual accounts retain the explicitly selected projects."""
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    config = {
+        "connectionMethod": "manual",
+        "deploymentProjectId": "signoz-test-project",
+        "deploymentRegion": "us-central1",
+        "projectIds": ["signoz-test-project", "signoz-test-project-2"],
+    }
+    account = create_cloud_integration_account(admin_token, "gcp", config={"gcp": config})
+    assert account["connectionArtifact"].get("gcp") is None
+
+    response = requests.get(
+        signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/gcp/accounts/{account['id']}"),
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=10,
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["config"]["gcp"] == config
+
+    checkin = simulate_agent_checkin(signoz, admin_token, "gcp", account["id"], "manual-account")
+    assert checkin.status_code == HTTPStatus.OK, checkin.text
+    assert checkin.json()["data"]["integrationConfig"] == {}
+
+
+@pytest.mark.parametrize(
+    "config_changes,omitted_field,error_field",
+    [
+        pytest.param({}, "connectionMethod", "connectionMethod", id="missing_connection_method"),
+        pytest.param({"connectionMethod": None}, None, "connectionMethod", id="null_connection_method"),
+        pytest.param({"connectionMethod": ""}, None, "connectionMethod", id="empty_connection_method"),
+        pytest.param({"connectionMethod": "   "}, None, "connectionMethod", id="blank_connection_method"),
+        pytest.param({"connectionMethod": "unsupported"}, None, "connectionMethod", id="unsupported_connection_method"),
+        *[pytest.param({"connectionMethod": method}, "projectIds", "projectIds", id=f"{method}_missing_project_ids") for method in ("manual", "agent")],
+        *[
+            pytest.param({"connectionMethod": method, "projectIds": project_ids}, None, "projectIds", id=f"{method}_{case}")
+            for method in ("manual", "agent")
+            for case, project_ids in (
+                ("null_project_ids", None),
+                ("empty_project_ids", []),
+                ("empty_project_id", [""]),
+                ("blank_project_id", ["   "]),
+                ("padded_project_id", [" signoz-validation-project "]),
+            )
+        ],
+        pytest.param({"connectionMethod": "agent", "projectIds": ["signoz-other-project"]}, None, "projectIds", id="agent_mismatched_project"),
+        pytest.param({"connectionMethod": "agent", "projectIds": ["signoz-validation-project", "signoz-other-project"]}, None, "projectIds", id="agent_multiple_projects"),
+        pytest.param({"deploymentProjectId": " signoz-validation-project "}, None, "deploymentProjectId", id="padded_deployment_project"),
+    ],
+)
+def test_create_gcp_account_rejects_invalid_config(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    config_changes: dict,
+    omitted_field: str | None,
+    error_field: str,
+) -> None:
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    config = {
+        "connectionMethod": "manual",
+        "deploymentProjectId": "signoz-validation-project",
+        "deploymentRegion": "us-central1",
+        "projectIds": ["signoz-validation-project"],
+    } | config_changes
+    if omitted_field is not None:
+        del config[omitted_field]
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/cloud_integrations/gcp/accounts"),
+        headers={"Authorization": f"Bearer {admin_token}"},
+        json={
+            "config": {"gcp": config},
+            "credentials": {
+                "sigNozApiUrl": "https://test-deployment.test.signoz.cloud",
+                "sigNozApiKey": "test-api-key-789",
+                "ingestionUrl": "https://ingest.test.signoz.cloud",
+                "ingestionKey": "test-ingestion-key-123456",
+            },
+        },
+        timeout=10,
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert error_field in response.text
+
+
+@pytest.mark.parametrize(
+    "secret_versions,expected_api_key_version,expected_ingestion_key_version",
+    [
+        pytest.param({"api_key_version": "3", "ingestion_key_version": "7"}, "3", "7", id="uploaded_versions"),
+        pytest.param({}, "latest", "latest", id="latest_versions"),
+    ],
+)
+def test_create_gcp_agent_account_connection_command(
+    create_user_admin: types.Operation,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    create_cloud_integration_account: Callable,
+    secret_versions: dict,
+    expected_api_key_version: str,
+    expected_ingestion_key_version: str,
+) -> None:
+    """The generated apply command uses the bootstrap's secret references and the uploaded versions, defaulting to latest."""
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    account = create_cloud_integration_account(
+        admin_token,
+        "gcp",
+        config={"gcp": {"connectionMethod": "agent", "deploymentProjectId": "signoz-agent-project", "deploymentRegion": "us-central1", "projectIds": ["signoz-agent-project"]}},
+    )
+    artifact = account["connectionArtifact"]["gcp"]
+    command = artifact["cliCommand"]
+    service_account_command = artifact["createServiceAccountCommand"]
+    secrets_command = artifact["createSecretsCommand"]
+
+    assert "gcloud iam service-accounts create 'signoz-integration-bootstrap'" in service_account_command
+    assert "executor_email='signoz-integration-bootstrap@signoz-agent-project.iam.gserviceaccount.com'" in service_account_command
+    assert "gcloud config get-value account" in service_account_command
+    assert "roles/config.agent" in service_account_command
+    assert "--role=roles/secretmanager.admin" in service_account_command
+
+    for prerequisite_command in (service_account_command, secrets_command):
+        assert "project_id='signoz-agent-project'" in prerequisite_command
+        assert "api_secret_id='signoz-integration-credentials-api-key'" in prerequisite_command
+        assert "ingestion_secret_id='signoz-integration-credentials-ingestion-key'" in prerequisite_command
+
+    assert "region='us-central1'" in secrets_command
+    assert 'gcloud secrets create "$api_secret_id"' in secrets_command
+    assert 'gcloud secrets create "$ingestion_secret_id"' in secrets_command
+    assert "--replication-policy=user-managed" in secrets_command
+    assert "--labels=managed-by=signoz-integration" in secrets_command
+    assert "printf '%s' 'test-api-key-789'" in secrets_command
+    assert "printf '%s' 'test-ingestion-key-123456'" in secrets_command
+
+    # Capture the expanded arguments with a shell function instead of deploying GCP resources.
+    script = "gcloud() { printf '%s\\0' \"$@\"; }\n" + command
+    env = {key: value for key, value in os.environ.items() if key not in ("api_key_version", "ingestion_key_version")} | secret_versions
+    result = subprocess.run(["/bin/sh", "-c", script], env=env, capture_output=True, text=True, check=True, timeout=5)
+    args = result.stdout.split("\0")[:-1]
+    assert args[:4] == ["infra-manager", "deployments", "apply", "projects/signoz-agent-project/locations/us-central1/deployments/signoz-integration-bootstrap"]
+    flags = dict(arg.removeprefix("--").split("=", 1) for arg in args[4:])
+    assert flags["project"] == "signoz-agent-project"
+    assert flags["service-account"] == "projects/signoz-agent-project/serviceAccounts/signoz-integration-bootstrap@signoz-agent-project.iam.gserviceaccount.com"
+    assert flags["gcs-source"] == "gs://signoz-cloud-integration/gcp/v0.0.8/bootstrap/"
+    assert flags["tf-version-constraint"] == "=1.5.7"
+    assert dict(value.split("=", 1) for value in flags["input-values"].split(",")) == {
+        "project_id": "signoz-agent-project",
+        "deployment_region": "us-central1",
+        "integration_account_id": account["id"],
+        "signoz_api_url": "https://test-deployment.test.signoz.cloud",
+        "ingestion_url": "https://ingest.test.signoz.cloud",
+        "api_key_secret_id": "signoz-integration-credentials-api-key",
+        "api_key_secret_version": expected_api_key_version,
+        "ingestion_key_secret_id": "signoz-integration-credentials-ingestion-key",
+        "ingestion_key_secret_version": expected_ingestion_key_version,
+    }
+    assert "test-api-key-789" not in command
+    assert "test-ingestion-key-123456" not in command
+
+
 def test_create_account_unsupported_provider(
     signoz: types.SigNoz,
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
@@ -135,6 +300,7 @@ def test_create_gcp_account_without_project_ids(
         json={
             "config": {
                 "gcp": {
+                    "connectionMethod": "manual",
                     "deploymentProjectId": "signoz-test-project",
                     "deploymentRegion": "us-central1",
                     "projectIds": [],

tests/integration/tests/cloudintegrations/03_agent_check_in.py

@@ -31,12 +31,14 @@ GCP_ACCOUNT_SPEC = ProviderAccountSpec(
     },
     build_config=lambda p: {
         "gcp": {
+            "connectionMethod": "manual",
             "deploymentProjectId": p["deployment_project_id"],
             "deploymentRegion": p["deployment_region"],
             "projectIds": p["project_ids"],
         }
     },
     expected_config=lambda p: {
+        "connectionMethod": "manual",
         "deploymentProjectId": p["deployment_project_id"],
         "deploymentRegion": p["deployment_region"],
         "projectIds": p["project_ids"],
@@ -51,6 +53,25 @@ provider_spec = pytest.mark.parametrize(
     ids=[s.id for s in PROVIDER_ACCOUNT_SPECS],
 )
 
+provider_account_reuse = pytest.mark.parametrize(
+    "provider,account_config,provider_account_id",
+    [
+        pytest.param("aws", {"aws": {"deploymentRegion": "us-east-1", "regions": ["us-east-1"]}}, "123456789012", id="aws"),
+        pytest.param(
+            "gcp",
+            {"gcp": {"connectionMethod": "manual", "deploymentProjectId": "signoz-reuse-project", "deploymentRegion": "us-central1", "projectIds": ["signoz-reuse-project"]}},
+            "reused-manual-account",
+            id="gcp_manual",
+        ),
+        pytest.param(
+            "gcp",
+            {"gcp": {"connectionMethod": "agent", "deploymentProjectId": "signoz-reuse-project", "deploymentRegion": "us-central1", "projectIds": ["signoz-reuse-project"]}},
+            "signoz-reuse-project",
+            id="gcp_agent",
+        ),
+    ],
+)
+
 
 def test_apply_license(
     signoz: types.SigNoz,
@@ -155,6 +176,201 @@ def test_duplicate_cloud_account_checkins(
     assert response.status_code == HTTPStatus.CONFLICT, f"Expected 409 for duplicate providerAccountId, got {response.status_code}: {response.text}"
 
 
+def test_gcp_agent_check_in_returns_collection_strategy(
+    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)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    project_id = "signoz-agent-strategy-project"
+    account_id = create_cloud_integration_account(admin_token, "gcp", config={"gcp": {"connectionMethod": "agent", "deploymentProjectId": project_id, "deploymentRegion": "us-central1", "projectIds": [project_id]}})["id"]
+    services_url = signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/gcp/accounts/{account_id}/services")
+    mysql_sink_filter = '(resource.type="cloudsql_database" AND (logName:"/logs/cloudsql.googleapis.com%2Fmysql" OR (log_id("cloudaudit.googleapis.com/data_access") AND protoPayload.request.@type="type.googleapis.com/google.cloud.sql.audit.v1.MysqlAuditEntry")))'
+    postgres_sink_filter = '(resource.type="cloudsql_database" AND (logName:"/logs/cloudsql.googleapis.com%2Fpostgres" OR (log_id("cloudaudit.googleapis.com/data_access") AND protoPayload.request.@type="type.googleapis.com/google.cloud.sql.audit.v1.PgAuditEntry")))'
+
+    response = simulate_agent_checkin(signoz, admin_token, "gcp", account_id, project_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["integrationConfig"]["gcp"] == {
+        "connectionMethod": "agent",
+        "deploymentProjectId": project_id,
+        "deploymentRegion": "us-central1",
+        "projectIds": [project_id],
+        "telemetryCollectionStrategy": {},
+    }
+
+    for service_id, service_config in (
+        ("cloudsql_mysql", {"metrics": {"enabled": True}, "logs": {"enabled": False}}),
+        ("cloudsql_postgres", {"metrics": {"enabled": True}, "logs": {"enabled": True}}),
+    ):
+        response = requests.put(f"{services_url}/{service_id}", headers=headers, json={"config": {"gcp": service_config}}, timeout=10)
+        assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    response = simulate_agent_checkin(signoz, admin_token, "gcp", account_id, project_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+    strategy = response.json()["data"]["integrationConfig"]["gcp"]["telemetryCollectionStrategy"]
+    metric_names = strategy["metrics"]["metricNames"]
+    assert metric_names == sorted(set(metric_names)), "metrics shared by MySQL and Postgres should be listed once, sorted"
+    assert {"cloudsql.googleapis.com/database/up", "cloudsql.googleapis.com/database/mysql/queries", "cloudsql.googleapis.com/database/postgresql/num_backends"} <= set(metric_names)
+    assert strategy["logs"] == {"sinkFilter": postgres_sink_filter}, "only services with logs enabled should contribute a sink filter"
+
+    response = requests.put(f"{services_url}/cloudsql_mysql", headers=headers, json={"config": {"gcp": {"metrics": {"enabled": False}, "logs": {"enabled": True}}}}, timeout=10)
+    assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    response = simulate_agent_checkin(signoz, admin_token, "gcp", account_id, project_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+    strategy = response.json()["data"]["integrationConfig"]["gcp"]["telemetryCollectionStrategy"]
+    assert "cloudsql.googleapis.com/database/up" in strategy["metrics"]["metricNames"], "Postgres should still collect the shared metrics"
+    assert not [name for name in strategy["metrics"]["metricNames"] if "/database/mysql/" in name], "MySQL metrics should stop once its metrics are disabled"
+    assert strategy["logs"] == {"sinkFilter": f"{mysql_sink_filter} OR {postgres_sink_filter}"}
+
+
+def test_gcp_agent_check_in_rejects_other_project(
+    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)
+    project_id = "signoz-agent-mismatch-project"
+    account_id = create_cloud_integration_account(admin_token, "gcp", config={"gcp": {"connectionMethod": "agent", "deploymentProjectId": project_id, "deploymentRegion": "us-central1", "projectIds": [project_id]}})["id"]
+
+    response = simulate_agent_checkin(signoz, admin_token, "gcp", account_id, "signoz-other-project")
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+
+    response = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/gcp/accounts/{account_id}"), headers={"Authorization": f"Bearer {admin_token}"}, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["providerAccountId"] is None, "a rejected check-in must not connect the account"
+    assert response.json()["data"]["agentReport"] is None
+
+    response = simulate_agent_checkin(signoz, admin_token, "gcp", account_id, project_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+
+
+@provider_account_reuse
+def test_removed_account_checkins_after_provider_account_id_is_reused(
+    signoz: types.SigNoz,
+    create_user_admin: types.Operation,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    create_cloud_integration_account: Callable,
+    provider: str,
+    account_config: dict,
+    provider_account_id: str,
+) -> None:
+    """Connect old account, remove it, connect replacement, then check in from the old agent."""
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    accounts_url = signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/{provider}/accounts")
+
+    # The original agent connects before its integration is removed.
+    old_account_id = create_cloud_integration_account(admin_token, provider, config=account_config)["id"]
+    response = simulate_agent_checkin(signoz, admin_token, provider, old_account_id, provider_account_id, data={"version": "old-agent"})
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    response = requests.delete(f"{accounts_url}/{old_account_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    response = requests.get(f"{accounts_url}/{old_account_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    old_account_before_checkins = response.json()["data"]
+
+    # A new integration claims the same provider account ID.
+    replacement_id = create_cloud_integration_account(admin_token, provider, config=account_config)["id"]
+    response = simulate_agent_checkin(signoz, admin_token, provider, replacement_id, provider_account_id, data={"version": "replacement-agent"})
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    response = requests.get(f"{accounts_url}/{replacement_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    replacement_before_checkins = response.json()["data"]
+
+    # The old agent keeps retrying. It must receive removal instructions, rather than a 409.
+    for _ in range(3):
+        response = simulate_agent_checkin(signoz, admin_token, provider, old_account_id, provider_account_id, data={"version": "old-agent-after-removal"})
+        assert response.status_code == HTTPStatus.OK, response.text
+        cleanup_response = response.json()["data"]
+        assert cleanup_response["cloudIntegrationId"] == old_account_id
+        assert cleanup_response["providerAccountId"] == provider_account_id
+        assert cleanup_response["removedAt"] is not None
+        assert cleanup_response["removedAt"] == old_account_before_checkins["removedAt"]
+        assert cleanup_response["integrationConfig"] == {}
+
+    # Cleanup may update sync state, but must preserve the original heartbeat, data and identity.
+    response = requests.get(f"{accounts_url}/{old_account_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    removed_account = response.json()["data"]
+    assert removed_account["providerAccountId"] == old_account_before_checkins["providerAccountId"]
+    assert removed_account["agentReport"]["timestampMillis"] == old_account_before_checkins["agentReport"]["timestampMillis"]
+    assert removed_account["agentReport"]["data"] == old_account_before_checkins["agentReport"]["data"]
+    assert removed_account["agentReport"]["syncState"] == cleanup_response["syncState"]
+
+    response = requests.get(f"{accounts_url}/{replacement_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"] == replacement_before_checkins, "old check-ins must not change the replacement account"
+
+    response = simulate_agent_checkin(signoz, admin_token, provider, replacement_id, provider_account_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["removedAt"] is None
+
+
+@provider_account_reuse
+def test_removed_account_first_checkin_after_provider_account_id_is_reused(
+    signoz: types.SigNoz,
+    create_user_admin: types.Operation,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    create_cloud_integration_account: Callable,
+    provider: str,
+    account_config: dict,
+    provider_account_id: str,
+) -> None:
+    """Removal before the first heartbeat must still be reported after another integration claims the provider ID."""
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    accounts_url = signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/{provider}/accounts")
+
+    # Remove the original integration before its agent has checked in even once.
+    old_account_id = create_cloud_integration_account(admin_token, provider, config=account_config)["id"]
+    response = requests.delete(f"{accounts_url}/{old_account_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    response = requests.get(f"{accounts_url}/{old_account_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    old_account_before_checkins = response.json()["data"]
+    assert old_account_before_checkins["providerAccountId"] is None
+    assert old_account_before_checkins["agentReport"] is None
+
+    replacement_id = create_cloud_integration_account(admin_token, provider, config=account_config)["id"]
+    response = simulate_agent_checkin(signoz, admin_token, provider, replacement_id, provider_account_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    response = requests.get(f"{accounts_url}/{replacement_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    replacement_before_checkins = response.json()["data"]
+
+    # Late first check-ins must report removal without creating a heartbeat or claiming the ID.
+    for _ in range(3):
+        response = simulate_agent_checkin(signoz, admin_token, provider, old_account_id, provider_account_id)
+        assert response.status_code == HTTPStatus.OK, response.text
+        cleanup_response = response.json()["data"]
+        assert cleanup_response["cloudIntegrationId"] == old_account_id
+        assert cleanup_response["providerAccountId"] == provider_account_id
+        assert cleanup_response["removedAt"] == old_account_before_checkins["removedAt"]
+        assert cleanup_response["removedAt"] is not None
+        assert cleanup_response["integrationConfig"] == {}
+
+    response = requests.get(f"{accounts_url}/{old_account_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"] == old_account_before_checkins, "late first check-ins must not change a removed account"
+
+    response = requests.get(f"{accounts_url}/{replacement_id}", headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"] == replacement_before_checkins
+
+    response = simulate_agent_checkin(signoz, admin_token, provider, replacement_id, provider_account_id)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["removedAt"] is None
+
+
 def test_sync_state_drops_removed_region_after_ack(
     signoz: types.SigNoz,
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
@@ -346,12 +562,13 @@ def test_sync_state_region_removed_during_sync(
     }
 
 
-def test_sync_state_after_disconnect(
+def test_sync_state_after_disconnect_when_provider_account_id_is_reused(
     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 old AWS agent can finish cleanup while its provider ID belongs to a replacement integration."""
     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())
@@ -366,6 +583,10 @@ def test_sync_state_after_disconnect(
     )
     assert response.status_code == HTTPStatus.NO_CONTENT, response.text
 
+    replacement_id = create_cloud_integration_account(admin_token, "aws", regions=["us-east-1", "us-west-2"])["id"]
+    response = simulate_agent_checkin(signoz, admin_token, "aws", replacement_id, provider_account_id)
+    assert response.status_code == HTTPStatus.OK, 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
@@ -380,3 +601,11 @@ def test_sync_state_after_disconnect(
         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"
+
+    response = requests.get(
+        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.OK, response.text
+    assert response.json()["data"]["agentReport"]["syncState"] == {"version": 2, "inSync": True, "regions": {}}, "cleanup acknowledgment must be saved"

tests/integration/tests/cloudintegrations/04_accounts.py

@@ -1,9 +1,11 @@
+import json
 import uuid
 from collections.abc import Callable
 from http import HTTPStatus
 
 import pytest
 import requests
+from sqlalchemy import sql
 
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD, add_license
@@ -42,12 +44,14 @@ GCP_ACCOUNT_SPEC = ProviderAccountSpec(
     },
     build_config=lambda p: {
         "gcp": {
+            "connectionMethod": "manual",
             "deploymentProjectId": p["deployment_project_id"],
             "deploymentRegion": p["deployment_region"],
             "projectIds": p["project_ids"],
         }
     },
     expected_config=lambda p: {
+        "connectionMethod": "manual",
         "deploymentProjectId": p["deployment_project_id"],
         "deploymentRegion": p["deployment_region"],
         "projectIds": p["project_ids"],
@@ -73,6 +77,47 @@ def test_apply_license(
     add_license(signoz, make_http_mocks, get_token)
 
 
+def test_gcp_account_without_connection_method_remains_manual(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_cloud_integration_account: Callable,
+) -> None:
+    """A stored manual account without the new field can still be read, connected and updated."""
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    project_id = "signoz-test-project"
+    config = {"deploymentProjectId": project_id, "deploymentRegion": "us-central1", "projectIds": [project_id]}
+    account = create_cloud_integration_account(admin_token, "gcp", config={"gcp": config | {"connectionMethod": "manual"}})
+
+    # Reproduce the released manual account format, which had no connection method.
+    with signoz.sqlstore.conn.begin() as conn:
+        conn.execute(
+            sql.text("UPDATE cloud_integration SET config=:config WHERE id=:id"),
+            {"config": json.dumps(config), "id": account["id"]},
+        )
+
+    account_url = signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/gcp/accounts/{account['id']}")
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    response = requests.get(
+        account_url,
+        headers=headers,
+        timeout=10,
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["config"]["gcp"] == config | {"connectionMethod": "manual"}
+
+    checkin = simulate_agent_checkin(signoz, admin_token, "gcp", account["id"], project_id)
+    assert checkin.status_code == HTTPStatus.OK, checkin.text
+    assert checkin.json()["data"]["integrationConfig"] == {}
+
+    updated_config = config | {"projectIds": [project_id, "signoz-test-project-2"]}
+    response = requests.put(account_url, headers=headers, json={"config": {"gcp": updated_config}}, timeout=10)
+    assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    with signoz.sqlstore.conn.connect() as conn:
+        stored = conn.execute(sql.text("SELECT config FROM cloud_integration WHERE id=:id"), {"id": account["id"]}).scalar_one()
+    assert json.loads(stored) == updated_config | {"connectionMethod": "manual"}
+
+
 @provider_spec
 def test_list_accounts_empty(
     signoz: types.SigNoz,
@@ -293,6 +338,34 @@ def test_update_account_after_checkin_preserves_connected_status(
     assert found_after["removedAt"] is None, "removedAt should still be null"
 
 
+def test_update_gcp_agent_account_rejected(
+    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)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    project_id = "signoz-agent-update-project"
+    account_id = create_cloud_integration_account(admin_token, "gcp", config={"gcp": {"connectionMethod": "agent", "deploymentProjectId": project_id, "deploymentRegion": "us-central1", "projectIds": [project_id]}})["id"]
+    account_url = signoz.self.host_configs["8080"].get(f"/api/v1/cloud_integrations/gcp/accounts/{account_id}")
+
+    checkin = simulate_agent_checkin(signoz, admin_token, "gcp", account_id, project_id)
+    assert checkin.status_code == HTTPStatus.OK, checkin.text
+
+    response = requests.put(account_url, headers=headers, json={"config": {"gcp": {"deploymentProjectId": project_id, "deploymentRegion": "us-central1", "projectIds": [project_id, "signoz-agent-update-project-2"]}}}, timeout=10)
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+
+    response = requests.get(account_url, headers=headers, timeout=10)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["data"]["config"]["gcp"] == {
+        "connectionMethod": "agent",
+        "deploymentProjectId": project_id,
+        "deploymentRegion": "us-central1",
+        "projectIds": [project_id],
+    }, "a rejected update must not change the agent account config"
+
+
 @provider_spec
 def test_disconnect_account(
     signoz: types.SigNoz,

tests/integration/tests/cloudintegrations/05_services.py

@@ -26,9 +26,10 @@ AWS_SERVICE_SPEC = ProviderServiceSpec(
 GCP_SERVICE_SPEC = ProviderServiceSpec(
     provider="gcp",
     service_id="cloudsql_postgres",
-    supports_logs=False,
+    supports_logs=True,
     account_config={
         "gcp": {
+            "connectionMethod": "manual",
             "deploymentProjectId": "signoz-test-project",
             "deploymentRegion": "us-central1",
             "projectIds": ["signoz-test-project"],
@@ -337,7 +338,7 @@ def test_update_gcp_service_without_metrics_config(
     get_token: Callable[[str, str], str],
     create_cloud_integration_account: Callable,
 ) -> None:
-    """GCP services support metrics only, so a config omitting metrics is rejected."""
+    """GCP services require a metrics config, so a logs-only config is rejected."""
     admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     account = create_cloud_integration_account(admin_token, "gcp", config=GCP_SERVICE_SPEC.account_config)

tests/integration/tests/tracedetail/02_thread.py

@@ -34,7 +34,14 @@ def test_thread_returns_message_spans_in_order(
                 timestamp=now - timedelta(seconds=8), trace_id=trace_id, span_id=first_llm_id, parent_span_id=root_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": first_input, "gen_ai.output.messages": first_output}, attribute_write_mode="json_only"
             ),
             Traces(
-                timestamp=now - timedelta(seconds=6), trace_id=trace_id, span_id=tool_id, parent_span_id=root_id, name="execute_tool get_weather", resources=resources, attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.call.id": "call_1", "gen_ai.tool.call.arguments": json.dumps({"city": "Bangalore"}), "gen_ai.tool.call.result": "sunny"}, attribute_write_mode="json_only"
+                timestamp=now - timedelta(seconds=6),
+                trace_id=trace_id,
+                span_id=tool_id,
+                parent_span_id=root_id,
+                name="execute_tool get_weather",
+                resources=resources,
+                attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.call.id": "call_1", "gen_ai.tool.call.arguments": json.dumps({"city": "Bangalore"}), "gen_ai.tool.call.result": "sunny"},
+                attribute_write_mode="json_only",
             ),
             Traces(timestamp=now - timedelta(seconds=5), trace_id=trace_id, span_id=agent_id, parent_span_id=root_id, name="invoke_agent planner", resources=resources, attributes={"gen_ai.agent.name": "planner"}, attribute_write_mode="json_only"),
             Traces(timestamp=now - timedelta(seconds=4), trace_id=trace_id, span_id=second_llm_id, parent_span_id=root_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": second_input}, attribute_write_mode="json_only"),