Coletar registros do Sentry

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Este documento explica como ingerir registros do Sentry no Google Security Operations usando o Google Cloud Storage. O Sentry gera dados operacionais na forma de eventos, problemas, dados de monitoramento de desempenho e informações de rastreamento de erros. Com essa integração, é possível enviar esses registros ao Google SecOps para análise e monitoramento, oferecendo visibilidade de erros de aplicativos, problemas de desempenho e interações do usuário nos aplicativos monitorados pelo Sentry.

Antes de começar

Verifique se você atende os seguintes pré-requisitos:

• Uma instância do Google SecOps
• Um projeto do GCP com a API Cloud Storage ativada
• Permissões para criar e gerenciar buckets do GCS
• Permissões para gerenciar políticas do IAM em buckets do GCS
• Permissões para criar funções do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
• Acesso privilegiado ao locatário do Sentry (token de autenticação com escopos de API)

Coletar pré-requisitos do Sentry (IDs, chaves de API, IDs de organização, tokens)

1. Faça login no Sentry.
2. Encontre o slug da organização:
   • Acesse Configurações > Organização > Configurações > ID da organização. O slug aparece ao lado do nome da organização.
3. Crie um token de autenticação:
   1. Acesse Configurações > Configurações do desenvolvedor > Tokens pessoais.
   2. Clique em Criar novo token.
   3. Escopos (mínimo): org:read, project:read, event:read.
   4. Clique em Criar token.
   5. Copie o valor do token (mostrado uma vez). Isso é usado como: Authorization: Bearer <token>.

4. Se você tiver hospedagem própria, anote o URL base (por exemplo, https://<your-domain>). Caso contrário, use https://sentry.io.

   Observação: o Sentry limita a taxa de todas as solicitações de API para evitar abusos. Os limites são aplicados a cada combinação exclusiva de chamador e endpoint, e cada endpoint tem um número máximo de solicitações e um tamanho de janela próprios. A função do Cloud Run implementa o backoff exponencial para lidar com a limitação de taxa automaticamente.
   Observação: os registros de erros são armazenados por 30 dias nos planos Free e 30 dias nos planos Team/Business. O Sentry também retém intervalos e transações amostrados por 13 meses nos planos Business. Isso limita os dados históricos disponíveis para preenchimento quando a integração é configurada pela primeira vez.

Criar um bucket do Google Cloud Storage

1. Acesse o Console do Google Cloud.
2. Selecione seu projeto ou crie um novo.
3. No menu de navegação, acesse Cloud Storage > Buckets.
4. Clique em Criar bucket.

5. Informe os seguintes detalhes de configuração:

   Configuração	Valor
   Nomeie seu bucket	Insira um nome exclusivo globalmente, por exemplo, sentry-logs.
   Tipo de local	Escolha com base nas suas necessidades (região, birregional, multirregional)
   Local	Selecione o local (por exemplo, us-central1).
   Classe de armazenamento	Padrão (recomendado para registros acessados com frequência)
   Controle de acesso	Uniforme (recomendado)
   Ferramentas de proteção	Opcional: ativar o controle de versões de objetos ou a política de retenção

6. Clique em Criar.

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar no bucket do GCS e ser invocada pelo Pub/Sub.

Criar conta de serviço

1. No Console do GCP, acesse IAM e administrador > Contas de serviço.
2. Clique em Criar conta de serviço.
3. Informe os seguintes detalhes de configuração:
   • Nome da conta de serviço: insira sentry-logs-collector-sa.
   • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Sentry logs.
4. Clique em Criar e continuar.
5. Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
   1. Clique em Selecionar papel.
   2. Pesquise e selecione Administrador de objetos do Storage.
   3. Clique em + Adicionar outro papel.
   4. Pesquise e selecione Invocador do Cloud Run.
   5. Clique em + Adicionar outro papel.
   6. Pesquise e selecione Invocador do Cloud Functions.
6. Clique em Continuar.
7. Clique em Concluído.

Esses papéis são necessários para:

• Administrador de objetos do Storage: grava registros em um bucket do GCS e gerencia arquivos de estado.
• Invocador do Cloud Run: permite que o Pub/Sub invoque a função
• Invocador do Cloud Functions: permite a invocação de funções

Conceder permissões do IAM no bucket do GCS

Conceda permissões de gravação à conta de serviço no bucket do GCS:

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, sentry-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
   • Atribuir papéis: selecione Administrador de objetos do Storage.
6. Clique em Salvar.

Criar tópico Pub/Sub

Crie um tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.

1. No Console do GCP, acesse Pub/Sub > Tópicos.
2. Selecione Criar tópico.
3. Informe os seguintes detalhes de configuração:
   • ID do tópico: insira sentry-logs-trigger.
   • Não altere as outras configurações.
4. Clique em Criar.

Criar uma função do Cloud Run para coletar registros

A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API Sentry e gravá-los no GCS.

1. No console do GCP, acesse o Cloud Run.
2. Clique em Criar serviço.
3. Selecione Função (use um editor in-line para criar uma função).

4. Na seção Configurar, forneça os seguintes detalhes de configuração:

   Configuração	Valor
   Nome do serviço	sentry-logs-collector
   Região	Selecione a região que corresponde ao seu bucket do GCS (por exemplo, us-central1).
   Ambiente de execução	Selecione Python 3.12 ou uma versão mais recente.

5. Na seção Acionador (opcional):

   1. Clique em + Adicionar gatilho.
   2. Selecione Cloud Pub/Sub.
   3. Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico (sentry-logs-trigger).
   4. Clique em Salvar.

6. Na seção Autenticação:

   1. Selecione Exigir autenticação.
   2. Confira o Identity and Access Management (IAM).
   Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.

7. Role a tela para baixo e abra Contêineres, rede, segurança.

8. Acesse a guia Segurança:

   • Conta de serviço: selecione a conta de serviço (sentry-logs-collector-sa).

9. Acesse a guia Contêineres:

   1. Clique em Variáveis e secrets.
   2. Clique em + Adicionar variável para cada variável de ambiente:

   Nome da variável	Valor de exemplo	Descrição
   GCS_BUCKET	sentry-logs	Nome do bucket do GCS em que os dados serão armazenados.
   GCS_PREFIX	sentry/events/	Prefixo opcional do GCS (subpasta) para objetos.
   STATE_KEY	sentry/events/state.json	Chave opcional do arquivo de estado/checkpoint.
   SENTRY_ORG	your-org-slug	Slug da organização do Sentry.
   SENTRY_AUTH_TOKEN	sntrys_************************	Token de autenticação do Sentry com org:read, project:read, event:read.
   SENTRY_API_BASE	https://sentry.io	URL base da API Sentry (autohospedado: https://<your-domain>).
   MAX_PROJECTS	100	Número máximo de projetos a serem processados.
   MAX_PAGES_PER_PROJECT	5	Número máximo de páginas por projeto e execução.

10. Role a tela para baixo na guia Variáveis e secrets até Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos).

11. Acesse a guia Configurações em Contêineres:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.
    • Clique em Concluído.

12. Role até Ambiente de execução:

    • Selecione Padrão (recomendado).

13. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100 ou ajuste com base na carga esperada.

14. Clique em Criar.

15. Aguarde a criação do serviço (1 a 2 minutos).

16. Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

1. Insira main em Ponto de entrada da função.

2. No editor de código em linha, crie dois arquivos:

   • Primeiro arquivo: main.py::

    import functions_framework
    from google.cloud import storage
    import json
    import os
    import urllib3
    from datetime import datetime, timezone
    import time
   
    # Initialize HTTP client
    http = urllib3.PoolManager()
   
    # Initialize Storage client
    storage_client = storage.Client()
   
    @functions_framework.cloud_event
    def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch Sentry events and write to GCS.
   
        Args:
            cloud_event: CloudEvent object containing Pub/Sub message
        """
   
        # Get environment variables
        bucket_name = os.environ.get('GCS_BUCKET')
        prefix = os.environ.get('GCS_PREFIX', 'sentry/events/')
        state_key = os.environ.get('STATE_KEY', 'sentry/events/state.json')
        org = os.environ.get('SENTRY_ORG', '').strip()
        token = os.environ.get('SENTRY_AUTH_TOKEN', '').strip()
        api_base = os.environ.get('SENTRY_API_BASE', 'https://sentry.io').rstrip('/')
        max_projects = int(os.environ.get('MAX_PROJECTS', '100'))
        max_pages_per_project = int(os.environ.get('MAX_PAGES_PER_PROJECT', '5'))
   
        if not all([bucket_name, org, token]):
            print('Error: Missing required environment variables')
            return
   
        try:
            # Get GCS bucket
            bucket = storage_client.bucket(bucket_name)
   
            # Load state
            state = load_state(bucket, state_key)
            state.setdefault('projects', {})
   
            # Get list of projects
            projects = list_projects(api_base, org, token, max_projects)
            print(f'Found {len(projects)} projects')
   
            summary = []
   
            # Process each project
            for slug in projects:
                start_prev = state['projects'].get(slug, {}).get('prev_cursor')
                res = fetch_project_events(
                    api_base, org, token, slug, start_prev, 
                    max_pages_per_project, bucket, prefix
                )
   
                if res.get('store_prev_cursor'):
                    state['projects'][slug] = {'prev_cursor': res['store_prev_cursor']}
   
                summary.append(res)
   
            # Save state
            save_state(bucket, state_key, state)
   
            print(f'Successfully processed {len(projects)} projects')
            print(f'Summary: {json.dumps(summary)}')
   
        except Exception as e:
            print(f'Error processing logs: {str(e)}')
            raise
   
    def load_state(bucket, key):
        """Load state from GCS."""
        try:
            blob = bucket.blob(key)
            if blob.exists():
                state_data = blob.download_as_text()
                return json.loads(state_data) if state_data else {'projects': {}}
        except Exception as e:
            print(f'Warning: Could not load state: {str(e)}')
        return {'projects': {}}
   
    def save_state(bucket, key, state):
        """Save state to GCS."""
        try:
            blob = bucket.blob(key)
            blob.upload_from_string(
                json.dumps(state, separators=(',', ':')),
                content_type='application/json'
            )
        except Exception as e:
            print(f'Warning: Could not save state: {str(e)}')
   
    def sentry_request(api_base, token, path, params=None):
        """Make request to Sentry API."""
        url = f"{api_base}{path}"
        if params:
            url = f"{url}?{urllib3.request.urlencode(params)}"
   
        headers = {
            'Authorization': f'Bearer {token}',
            'Accept': 'application/json',
            'User-Agent': 'chronicle-gcs-sentry-function/1.0'
        }
   
        response = http.request('GET', url, headers=headers, timeout=60.0)
        data = json.loads(response.data.decode('utf-8'))
        link = response.headers.get('Link')
   
        return data, link
   
    def parse_link_header(link_header):
        """Parse Link header to extract cursors."""
        if not link_header:
            return None, False, None, False
   
        prev_cursor, next_cursor = None, None
        prev_more, next_more = False, False
   
        parts = [p.strip() for p in link_header.split(',')]
        for p in parts:
            if '<' not in p or '>' not in p:
                continue
   
            url = p.split('<', 1)[1].split('>', 1)[0]
            rel = 'previous' if 'rel="previous"' in p else ('next' if 'rel="next"' in p else None)
            has_more = 'results="true"' in p
   
            try:
                from urllib.parse import urlparse, parse_qs
                q = urlparse(url).query
                cur = parse_qs(q).get('cursor', [None])[0]
            except Exception:
                cur = None
   
            if rel == 'previous':
                prev_cursor, prev_more = cur, has_more
            elif rel == 'next':
                next_cursor, next_more = cur, has_more
   
        return prev_cursor, prev_more, next_cursor, next_more
   
    def write_page(bucket, prefix, project_slug, payload, page_idx):
        """Write page of events to GCS."""
        ts = time.gmtime()
        key = f"{prefix.rstrip('/')}/{time.strftime('%Y/%m/%d', ts)}/sentry-{project_slug}-{page_idx:05d}.json"
   
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(payload, separators=(',', ':')),
            content_type='application/json'
        )
   
        return key
   
    def list_projects(api_base, org, token, max_projects):
        """List Sentry projects."""
        projects, cursor = [], None
   
        while len(projects) < max_projects:
            params = {'cursor': cursor} if cursor else {}
            data, link = sentry_request(api_base, token, f'/api/0/organizations/{org}/projects/', params)
   
            for p in data:
                slug = p.get('slug')
                if slug:
                    projects.append(slug)
                if len(projects) >= max_projects:
                    break
   
            _, _, next_cursor, next_more = parse_link_header(link)
            cursor = next_cursor if next_more else None
            if not next_more:
                break
   
        return projects
   
    def fetch_project_events(api_base, org, token, project_slug, start_prev_cursor, max_pages, bucket, prefix):
        """Fetch events for a project."""
        pages = 0
        total = 0
        latest_prev_cursor_to_store = None
   
        def fetch_one(cursor):
            nonlocal pages, total, latest_prev_cursor_to_store
   
            params = {'cursor': cursor} if cursor else {}
            data, link = sentry_request(api_base, token, f'/api/0/projects/{org}/{project_slug}/events/', params)
   
            write_page(bucket, prefix, project_slug, data, pages)
            total += len(data) if isinstance(data, list) else 0
   
            prev_c, prev_more, next_c, next_more = parse_link_header(link)
            latest_prev_cursor_to_store = prev_c or latest_prev_cursor_to_store
            pages += 1
   
            return prev_c, prev_more, next_c, next_more
   
        if start_prev_cursor:
            # Poll new pages toward "previous" until no more
            cur = start_prev_cursor
            while pages < max_pages:
                prev_c, prev_more, _, _ = fetch_one(cur)
                if not prev_more:
                    break
                cur = prev_c
        else:
            # First run: start at newest, then backfill older pages
            prev_c, _, next_c, next_more = fetch_one(None)
            cur = next_c
            while next_more and pages < max_pages:
                _, _, next_c, next_more = fetch_one(cur)
                cur = next_c
   
        return {
            'project': project_slug,
            'pages': pages,
            'written': total,
            'store_prev_cursor': latest_prev_cursor_to_store
        }
    ```
   
   * Second file: **requirements.txt:**
   
   

   functions-framework3.* google-cloud-storage2.* urllib3>=2.0.0 ```

3. Clique em Implantar para salvar e implantar a função.

4. Aguarde a conclusão da implantação (2 a 3 minutos).

   Observação: a configuração do acionador do Pub/Sub cria automaticamente as assinaturas e permissões necessárias.

Criar o job do Cloud Scheduler

O Cloud Scheduler publica mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

1. No Console do GCP, acesse o Cloud Scheduler.
2. Clique em Criar job.

3. Informe os seguintes detalhes de configuração:

   Configuração	Valor
   Nome	sentry-logs-collector-hourly
   Região	Selecione a mesma região da função do Cloud Run
   Frequência	0 * * * * (a cada hora, na hora)
   Fuso horário	Selecione o fuso horário (UTC recomendado)
   Tipo de destino	Pub/Sub
   Tópico	Selecione o assunto (sentry-logs-trigger)
   Corpo da mensagem	{} (objeto JSON vazio)

4. Clique em Criar.

Opções de frequência de programação

• Escolha a frequência com base no volume de registros e nos requisitos de latência:

  Frequência	Expressão Cron	Caso de uso
  A cada 5 minutos	*/5 * * * *	Alto volume e baixa latência
  A cada 15 minutos	*/15 * * * *	Volume médio
  A cada hora	0 * * * *	Padrão (recomendado)
  A cada 6 horas	0 */6 * * *	Baixo volume, processamento em lote
  Diário	0 0 * * *	Coleta de dados históricos

Testar o job do programador

1. No console do Cloud Scheduler, encontre seu job.
2. Clique em Forçar execução para acionar manualmente.
3. Aguarde alguns segundos e acesse Cloud Run > Serviços > sentry-logs-collector > Registros.
4. Verifique se a função foi executada com sucesso.
5. Verifique o bucket do GCS para confirmar se os registros foram gravados.

Recuperar a conta de serviço do Google SecOps

O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

1. Acesse Configurações do SIEM > Feeds.
2. Clique em Adicionar novo feed.
3. Clique em Configurar um único feed.
4. No campo Nome do feed, insira um nome para o feed (por exemplo, Sentry Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Sentry como o Tipo de registro.

7. Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço é exibido, por exemplo:

   [e-mail]
   

8. Copie esse endereço de e-mail para usar na próxima etapa.

   Observação: cada instância do Google SecOps tem uma conta de serviço exclusiva. Não use contas de serviço de outras documentações ou exemplos.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa do papel de Leitor de objetos do Storage no seu bucket do GCS.

1. Acesse Cloud Storage > Buckets.
2. Clique no nome do bucket.
3. Acesse a guia Permissões.
4. Clique em Conceder acesso.
5. Informe os seguintes detalhes de configuração:
   • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
   • Atribuir papéis: selecione Leitor de objetos do Storage.

6. Clique em Salvar.

   Observação: se você planeja usar a opção de exclusão "Excluir arquivos transferidos" ou "Excluir arquivos transferidos e diretórios vazios", conceda o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Configurar um feed no Google SecOps para ingerir registros do Sentry

1. Acesse Configurações do SIEM > Feeds.
2. Clique em Adicionar novo feed.
3. Clique em Configurar um único feed.
4. No campo Nome do feed, insira um nome para o feed (por exemplo, Sentry Logs).
5. Selecione Google Cloud Storage V2 como o Tipo de origem.
6. Selecione Sentry como o Tipo de registro.
7. Clique em Próxima.

8. Especifique valores para os seguintes parâmetros de entrada:

   • URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:

     gs://sentry-logs/sentry/events/
     

     • Substitua:

       • sentry-logs: o nome do bucket do GCS.
       • sentry/events/: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).

     • Exemplos:

       • Bucket raiz: gs://company-logs/
       • Com prefixo: gs://company-logs/sentry-logs/
       • Com subpasta: gs://company-logs/sentry/events/
     Observação: sempre inclua a barra (/) no final do URI.

   • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

     • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
     • Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.

     • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.

       Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.

   • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.

   • Namespace do recurso: o namespace do recurso.

   • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.

9. Clique em Próxima.

10. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.