14900461121

Committed 08 May 2025 06:49AM UTC coverage: 76.227% (+0.1%) from 76.125%

Build # 14900461121

Build Type

Pull #3250

github

Committed by

moukoublen

Commit Message

log errors on env vars parsing

Pull Request Pull Request #3250: Increase gcp ListAssets timeout and page size

Run Details

57 of 67 new or added lines in 5 files covered. (85.07%)

15 existing lines in 2 files now uncovered.

9209 of 12081 relevant lines covered (76.23%)

16.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.27

/internal/resources/providers/gcplib/inventory/provider.go

// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package inventory

import (
        "context"
        "fmt"
        "strings"
        "sync"

        asset "cloud.google.com/go/asset/apiv1"
        "cloud.google.com/go/asset/apiv1/assetpb"
        "github.com/googleapis/gax-go/v2"
        "github.com/samber/lo"
        "google.golang.org/api/cloudresourcemanager/v3"
        "google.golang.org/api/iterator"
        "google.golang.org/api/option"
        "google.golang.org/protobuf/types/known/structpb"

        "github.com/elastic/cloudbeat/internal/config"
        "github.com/elastic/cloudbeat/internal/infra/clog"
        "github.com/elastic/cloudbeat/internal/resources/fetching"
        "github.com/elastic/cloudbeat/internal/resources/providers/gcplib/auth"
)

type Provider struct {
        log                       *clog.Logger
        config                    auth.GcpFactoryConfig
        inventory                 *AssetsInventoryWrapper
        crm                       *ResourceManagerWrapper
        cloudAccountMetadataCache *MapCache[*fetching.CloudAccountMetadata]
}

type AssetsInventoryWrapper struct {
        Close      func() error
        ListAssets func(ctx context.Context, req *assetpb.ListAssetsRequest, opts ...gax.CallOption) Iterator
}

type ResourceManagerWrapper struct {
        // returns project display name or an empty string
        getProjectDisplayName func(ctx context.Context, parent string) string

        // returns org display name or an empty string
        getOrganizationDisplayName func(ctx context.Context, parent string) string
}

type MonitoringAsset struct {
        CloudAccount *fetching.CloudAccountMetadata
        LogMetrics   []*ExtendedGcpAsset `json:"log_metrics,omitempty"`
        Alerts       []*ExtendedGcpAsset `json:"alerts,omitempty"`
}

type LoggingAsset struct {
        CloudAccount *fetching.CloudAccountMetadata
        LogSinks     []*ExtendedGcpAsset `json:"log_sinks,omitempty"`
}

type ProjectPoliciesAsset struct {
        CloudAccount *fetching.CloudAccountMetadata
        Policies     []*ExtendedGcpAsset `json:"policies,omitempty"`
}

type ServiceUsageAsset struct {
        CloudAccount *fetching.CloudAccountMetadata
        Services     []*ExtendedGcpAsset `json:"services,omitempty"`
}

type ExtendedGcpAsset struct {
        *assetpb.Asset
        CloudAccount *fetching.CloudAccountMetadata
}

type ProviderInitializer struct{}

type dnsPolicyFields struct {
        networks      []string
        enableLogging bool
}

type TypeGenerator[T any] func(assets []*ExtendedGcpAsset, projectId, projectName, orgId, orgName string) *T

type Iterator interface {
        Next() (*assetpb.Asset, error)
}

type ServiceAPI interface {
        // ListAllAssetTypesByName List all content types of the given assets types
        ListAllAssetTypesByName(ctx context.Context, assets []string) ([]*ExtendedGcpAsset, error)

        // ListMonitoringAssets List all monitoring assets by project id
        ListMonitoringAssets(ctx context.Context, monitoringAssetTypes map[string][]string) ([]*MonitoringAsset, error)

        // ListLoggingAssets returns a list of logging assets grouped by project id, extended with folder and org level log sinks
        ListLoggingAssets(ctx context.Context) ([]*LoggingAsset, error)

        // ListServiceUsageAssets returns a list of service usage assets grouped by project id
        ListServiceUsageAssets(ctx context.Context) ([]*ServiceUsageAsset, error)

        // returns a project policies for all its ancestors
        ListProjectsAncestorsPolicies(ctx context.Context) ([]*ProjectPoliciesAsset, error)

        // Close the GCP asset client
        Close() error
}

type ProviderInitializerAPI interface {
        // Init initializes the GCP asset client
        Init(ctx context.Context, log *clog.Logger, gcpConfig auth.GcpFactoryConfig, cfg config.GcpConfig) (ServiceAPI, error)
}

func (p *ProviderInitializer) Init(ctx context.Context, log *clog.Logger, gcpConfig auth.GcpFactoryConfig, cfg config.GcpConfig) (ServiceAPI, error) {
        limiter := NewAssetsInventoryRateLimiter(log)
        // initialize GCP assets inventory client
        client, err := asset.NewClient(ctx, append(gcpConfig.ClientOpts, option.WithGRPCDialOption(limiter.GetInterceptorDialOption()))...)
        if err != nil {
                return nil, err
        }
        // wrap the assets inventory client for mocking
        assetsInventoryWrapper := &AssetsInventoryWrapper{
                Close: client.Close,
                ListAssets: func(ctx context.Context, req *assetpb.ListAssetsRequest, opts ...gax.CallOption) Iterator {
                        if req.PageSize == 0 {
                                req.PageSize = cfg.GcpCallOpt.ListAssetsPageSize
                        }
                        return client.ListAssets(ctx, req, append(opts, GAXCallOptionRetrier(log), gax.WithTimeout(cfg.GcpCallOpt.ListAssetsTimeout))...)
                },
        }

        // initialize GCP resource manager client
        var gcpClientOpt []option.ClientOption
        gcpClientOpt = append(append(gcpClientOpt, option.WithScopes(cloudresourcemanager.CloudPlatformReadOnlyScope)), gcpConfig.ClientOpts...)
        crmService, err := cloudresourcemanager.NewService(ctx, gcpClientOpt...)
        if err != nil {
                return nil, err
        }

        displayNamesCache := NewMapCache[string]()
        // wrap the resource manager client for mocking
        crmServiceWrapper := &ResourceManagerWrapper{
                getProjectDisplayName: func(ctx context.Context, parent string) string {
                        return displayNamesCache.Get(func() string {
                                prj, err := crmService.Projects.Get(parent).Context(ctx).Do()
                                if err != nil {
                                        log.Errorf("error fetching GCP Project: %s, error: %s", parent, err)
                                        return ""
                                }
                                return prj.DisplayName
                        }, parent)
                },
                getOrganizationDisplayName: func(ctx context.Context, parent string) string {
                        return displayNamesCache.Get(func() string {
                                org, err := crmService.Organizations.Get(parent).Context(ctx).Do()
                                if err != nil {
                                        log.Errorf("error fetching GCP Org: %s, error: %s", parent, err)
                                        return ""
                                }
                                return org.DisplayName
                        }, parent)
                },
        }

        return &Provider{
                config:                    gcpConfig,
                log:                       log,
                inventory:                 assetsInventoryWrapper,
                crm:                       crmServiceWrapper,
                cloudAccountMetadataCache: NewMapCache[*fetching.CloudAccountMetadata](),
        }, nil
}

func (p *Provider) ListAllAssetTypesByName(ctx context.Context, assetTypes []string) ([]*ExtendedGcpAsset, error) {
        wg := sync.WaitGroup{}
        var resourceAssets []*assetpb.Asset
        var policyAssets []*assetpb.Asset

        wg.Add(1)
        go func() {
                resourceAssets = p.getAllAssets(ctx, &assetpb.ListAssetsRequest{
                        Parent:      p.config.Parent,
                        AssetTypes:  assetTypes,
                        ContentType: assetpb.ContentType_RESOURCE,
                })
                wg.Done()
        }()
        wg.Add(1)
        go func() {
                policyAssets = p.getAllAssets(ctx, &assetpb.ListAssetsRequest{
                        Parent:      p.config.Parent,
                        AssetTypes:  assetTypes,
                        ContentType: assetpb.ContentType_IAM_POLICY,
                })
                wg.Done()
        }()

        wg.Wait()

        var assets []*assetpb.Asset
        assets = append(append(assets, resourceAssets...), policyAssets...)
        mergedAssets := mergeAssetContentType(assets)
        extendedAssets := p.extendWithCloudMetadata(ctx, mergedAssets)
        // Enrich network assets with dns policy
        p.enrichNetworkAssets(ctx, extendedAssets)

        return extendedAssets, nil
}

// ListMonitoringAssets returns a list of monitoring assets grouped by project id
func (p *Provider) ListMonitoringAssets(ctx context.Context, monitoringAssetTypes map[string][]string) ([]*MonitoringAsset, error) {
        logMetrics, err := p.ListAllAssetTypesByName(ctx, monitoringAssetTypes["LogMetric"])
        if err != nil {
                return nil, err
        }

        alertPolicies, err := p.ListAllAssetTypesByName(ctx, monitoringAssetTypes["AlertPolicy"])
        if err != nil {
                return nil, err
        }

        typeGenerator := func(assets []*ExtendedGcpAsset, projectId, projectName, orgId, orgName string) *MonitoringAsset {
                return &MonitoringAsset{
                        LogMetrics: getAssetsByType(assets, MonitoringLogMetricAssetType),
                        Alerts:     getAssetsByType(assets, MonitoringAlertPolicyAssetType),
                        CloudAccount: &fetching.CloudAccountMetadata{
                                AccountId:        projectId,
                                AccountName:      projectName,
                                OrganisationId:   orgId,
                                OrganizationName: orgName,
                        },
                }
        }

        var assets []*ExtendedGcpAsset
        assets = append(append(assets, logMetrics...), alertPolicies...)
        monitoringAssets := getAssetsByProject[MonitoringAsset](assets, p.log, typeGenerator)

        return monitoringAssets, nil
}

// ListLoggingAssets returns a list of logging assets grouped by project id, extended with folder and org level log sinks
func (p *Provider) ListLoggingAssets(ctx context.Context) ([]*LoggingAsset, error) {
        logSinks, err := p.ListAllAssetTypesByName(ctx, []string{LogSinkAssetType})
        if err != nil {
                return nil, err
        }

        typeGenerator := func(assets []*ExtendedGcpAsset, projectId, projectName, orgId, orgName string) *LoggingAsset {
                return &LoggingAsset{
                        LogSinks: assets,
                        CloudAccount: &fetching.CloudAccountMetadata{
                                AccountId:        projectId,
                                AccountName:      projectName,
                                OrganisationId:   orgId,
                                OrganizationName: orgName,
                        },
                }
        }

        loggingAssets := getAssetsByProject[LoggingAsset](logSinks, p.log, typeGenerator)
        return loggingAssets, nil
}

// ListServiceUsageAssets returns a list of service usage assets grouped by project id
func (p *Provider) ListServiceUsageAssets(ctx context.Context) ([]*ServiceUsageAsset, error) {
        services, err := p.ListAllAssetTypesByName(ctx, []string{ServiceUsageAssetType})
        if err != nil {
                return nil, err
        }

        typeGenerator := func(assets []*ExtendedGcpAsset, projectId, projectName, orgId, orgName string) *ServiceUsageAsset {
                return &ServiceUsageAsset{
                        Services: assets,
                        CloudAccount: &fetching.CloudAccountMetadata{
                                AccountId:        projectId,
                                AccountName:      projectName,
                                OrganisationId:   orgId,
                                OrganizationName: orgName,
                        },
                }
        }

        assets := getAssetsByProject[ServiceUsageAsset](services, p.log, typeGenerator)
        return assets, nil
}

func (p *Provider) Close() error {
        return p.inventory.Close()
}

// enrichNetworkAssets enriches the network assets with dns policy if exists
func (p *Provider) enrichNetworkAssets(ctx context.Context, assets []*ExtendedGcpAsset) {
        networkAssets := getAssetsByType(assets, ComputeNetworkAssetType)
        if len(networkAssets) == 0 {
                p.log.Infof("no %s assets were listed", ComputeNetworkAssetType)
                return
        }
        dnsPolicyAssets := p.getAllAssets(ctx, &assetpb.ListAssetsRequest{
                Parent:      p.config.Parent,
                AssetTypes:  []string{DnsPolicyAssetType},
                ContentType: assetpb.ContentType_RESOURCE,
        })

        if len(dnsPolicyAssets) == 0 {
                p.log.Infof("no %s assets were listed, return original assets", DnsPolicyAssetType)
                return
        }
        dnsPolicies := decodeDnsPolicies(dnsPolicyAssets)

        p.log.Infof("attempting to enrich %d %s assets with dns policy", len(assets), ComputeNetworkAssetType)
        for _, networkAsset := range networkAssets {
                networkAssetFields := networkAsset.GetResource().GetData().GetFields()
                networkIdentifier := strings.TrimPrefix(networkAsset.GetName(), "//compute.googleapis.com")

                dnsPolicy := findDnsPolicyByNetwork(dnsPolicies, networkIdentifier)
                if dnsPolicy != nil {
                        p.log.Infof("enrich a %s asset with dns policy, name: %s", ComputeNetworkAssetType, networkIdentifier)
                        networkAssetFields["enabledDnsLogging"] = &structpb.Value{Kind: &structpb.Value_BoolValue{BoolValue: dnsPolicy.enableLogging}}
                }
        }
}

// findDnsPolicyByNetwork finds DNS policy by network identifier
func findDnsPolicyByNetwork(dnsPolicies []*dnsPolicyFields, networkIdentifier string) *dnsPolicyFields {
        for _, dnsPolicy := range dnsPolicies {
                if lo.SomeBy(dnsPolicy.networks, func(networkUrl string) bool {
                        return strings.HasSuffix(networkUrl, networkIdentifier)
                }) {
                        return dnsPolicy
                }
        }
        return nil
}

// decodeDnsPolicies gets the required fields from the dns policies assets
func decodeDnsPolicies(dnsPolicyAssets []*assetpb.Asset) []*dnsPolicyFields {
        dnsPolicies := make([]*dnsPolicyFields, 0)
        for _, dnsPolicyAsset := range dnsPolicyAssets {
                fields := new(dnsPolicyFields)
                dnsPolicyData := dnsPolicyAsset.GetResource().GetData().GetFields()

                if attachedNetworks, exist := dnsPolicyData["networks"]; exist {
                        networks := attachedNetworks.GetListValue().GetValues()
                        for _, network := range networks {
                                if networkUrl, found := network.GetStructValue().GetFields()["networkUrl"]; found {
                                        fields.networks = append(fields.networks, networkUrl.GetStringValue())
                                }
                        }
                }

                if enableLogging, exist := dnsPolicyData["enableLogging"]; exist {
                        fields.enableLogging = enableLogging.GetBoolValue()
                }

                dnsPolicies = append(dnsPolicies, fields)
        }

        return dnsPolicies
}

// getAssetsByProject groups assets by project, extracts metadata for each project, and adds folder and organization-level resources for each group.
func getAssetsByProject[T any](assets []*ExtendedGcpAsset, log *clog.Logger, f TypeGenerator[T]) []*T {
        assetsByProject := lo.GroupBy(assets, func(asset *ExtendedGcpAsset) string { return asset.CloudAccount.AccountId })
        enrichedAssets := make([]*T, 0, len(assetsByProject))
        for projectId, projectAssets := range assetsByProject {
                if projectId == "" {
                        continue
                }

                if len(projectAssets) == 0 {
                        log.Errorf("no assets were listed for project: %s", projectId)
                        continue
                }

                cloudAccount := projectAssets[0].CloudAccount

                // add folder and org level log sinks for each project
                projectAssets = append(projectAssets, assetsByProject[""]...)
                enrichedAssets = append(enrichedAssets, f(
                        projectAssets,
                        projectId,
                        cloudAccount.AccountName,
                        cloudAccount.OrganisationId,
                        cloudAccount.OrganizationName,
                ))
        }
        return enrichedAssets
}

func (p *Provider) getAllAssets(ctx context.Context, request *assetpb.ListAssetsRequest) []*assetpb.Asset {
        p.log.Infof("Listing asset types: %v of type %v for %v", request.AssetTypes, request.ContentType, request.Parent)
        results := make([]*assetpb.Asset, 0)
        it := p.inventory.ListAssets(ctx, request)
        for {
                response, err := it.Next()
                if err == iterator.Done {
                        break
                }
                if err != nil {
                        p.log.Errorf("Error fetching GCP Asset: %s", err)
                        return results
                }

                p.log.Debugf("Fetched GCP Asset: %+v", response.Name)
                results = append(results, response)
        }
        return results
}

func mergeAssetContentType(assets []*assetpb.Asset) []*assetpb.Asset {
        resultsMap := make(map[string]*assetpb.Asset)
        for _, asset := range assets {
                assetKey := asset.Name
                if _, ok := resultsMap[assetKey]; !ok {
                        resultsMap[assetKey] = asset
                        continue
                }
                item := resultsMap[assetKey]
                if asset.Resource != nil {
                        item.Resource = asset.Resource
                }
                if asset.IamPolicy != nil {
                        item.IamPolicy = asset.IamPolicy
                }
        }
        results := make([]*assetpb.Asset, 0, len(resultsMap))
        for _, asset := range resultsMap {
                results = append(results, asset)
        }
        return results
}

// extends the assets with the project and organization display name
func (p *Provider) extendWithCloudMetadata(ctx context.Context, assets []*assetpb.Asset) []*ExtendedGcpAsset {
        extendedAssets := make([]*ExtendedGcpAsset, 0, len(assets))
        for _, asset := range assets {
                orgId := getOrganizationId(asset.Ancestors)
                projectId := getProjectId(asset.Ancestors)
                cacheKey := fmt.Sprintf("%s/%s", projectId, orgId)
                cloudAccount := p.cloudAccountMetadataCache.Get(func() *fetching.CloudAccountMetadata {
                        return p.getCloudAccountMetadata(ctx, projectId, orgId)
                }, cacheKey)
                extendedAssets = append(extendedAssets, &ExtendedGcpAsset{
                        Asset:        asset,
                        CloudAccount: cloudAccount,
                })
        }
        return extendedAssets
}

func (p *Provider) ListProjectsAncestorsPolicies(ctx context.Context) ([]*ProjectPoliciesAsset, error) {
        projects := p.getAllAssets(ctx, &assetpb.ListAssetsRequest{
                ContentType: assetpb.ContentType_IAM_POLICY,
                Parent:      p.config.Parent,
                AssetTypes:  []string{CrmProjectAssetType},
        })
        p.log.Infof("Listed %d GCP projects", len(projects))
        ancestorsPoliciesCache := NewMapCache[[]*ExtendedGcpAsset]()
        return lo.Map(projects, func(project *assetpb.Asset, _ int) *ProjectPoliciesAsset {
                projectAsset := p.extendWithCloudMetadata(ctx, []*assetpb.Asset{project})[0]
                // Skip first ancestor it as we already got it
                policiesAssets := append([]*ExtendedGcpAsset{projectAsset}, getAncestorsAssets(ctx, ancestorsPoliciesCache, p, project.Ancestors[1:])...)
                return &ProjectPoliciesAsset{CloudAccount: projectAsset.CloudAccount, Policies: policiesAssets}
        }), nil
}

func getAncestorsAssets(ctx context.Context, ancestorsPoliciesCache *MapCache[[]*ExtendedGcpAsset], p *Provider, ancestors []string) []*ExtendedGcpAsset {
        return lo.Flatten(lo.Map(ancestors, func(parent string, _ int) []*ExtendedGcpAsset {
                return ancestorsPoliciesCache.Get(func() []*ExtendedGcpAsset {
                        var assetType string
                        if isFolder(parent) {
                                assetType = CrmFolderAssetType
                        }
                        if isOrganization(parent) {
                                assetType = CrmOrgAssetType
                        }
                        return p.extendWithCloudMetadata(ctx, p.getAllAssets(ctx, &assetpb.ListAssetsRequest{
                                ContentType: assetpb.ContentType_IAM_POLICY,
                                Parent:      parent,
                                AssetTypes:  []string{assetType},
                        }))
                }, parent)
        }))
}

func (p *Provider) getCloudAccountMetadata(ctx context.Context, projectId string, orgId string) *fetching.CloudAccountMetadata {
        var orgName string
        var projectName string
        wg := sync.WaitGroup{}
        wg.Add(1)
        go func() {
                if isOrganization(p.config.Parent) {
                        orgName = p.crm.getOrganizationDisplayName(ctx, fmt.Sprintf("organizations/%s", orgId))
                }
                wg.Done()
        }()
        wg.Add(1)
        go func() {
                // some assets are not associated with a project
                if projectId != "" {
                        projectName = p.crm.getProjectDisplayName(ctx, fmt.Sprintf("projects/%s", projectId))
                }
                wg.Done()
        }()
        wg.Wait()
        return &fetching.CloudAccountMetadata{
                AccountId:        projectId,
                AccountName:      projectName,
                OrganisationId:   orgId,
                OrganizationName: orgName,
        }
}

func getOrganizationId(ancestors []string) string {
        last := ancestors[len(ancestors)-1]
        parts := strings.Split(last, "/") // organizations/1234567890

        if parts[0] == "organizations" {
                return parts[1]
        }

        return ""
}

func getProjectId(ancestors []string) string {
        parts := strings.Split(ancestors[0], "/") // projects/1234567890

        if parts[0] == "projects" {
                return parts[1]
        }

        return ""
}

func getAssetsByType(projectAssets []*ExtendedGcpAsset, assetType string) []*ExtendedGcpAsset {
        return lo.Filter(projectAssets, func(asset *ExtendedGcpAsset, _ int) bool {
                return asset.AssetType == assetType
        })
}

func isFolder(parent string) bool {
        return strings.HasPrefix(parent, "folders")
}

func isOrganization(parent string) bool {
        return strings.HasPrefix(parent, "organizations")
}

elastic / cloudbeat / 14900461121

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous