Skip streaming enqueue for denylisted audit event types

Huzaifa Iftikharrequested to merge
633500-skip-enqueue-for-denylisted-audit-event-types into master

What does this MR do and why?

A destination with a denylist filter for an event type still passes the streamable? enqueue gate, so AuditEvents::AuditEventStreamingWorker is enqueued for every such event and the event is only dropped at send time. This MR excludes denylisted destinations in allowed_for_event_type, so the enqueue gate matches the per-event allowed_to_stream? check and no job is enqueued.

Only group destinations support denylist filters today. Instance destinations are unchanged.

Query change (group destinations)

Adds one NOT EXISTS on matching deny filters. The subquery uses the existing unique index uniq_audit_group_event_filters_destination_id_and_event_type on (external_streaming_destination_id, audit_event_type).

-- Before
... AND ((NOT EXISTS (SELECT 1 FROM audit_events_group_streaming_event_type_filters f
                      WHERE f.external_streaming_destination_id = d.id AND f.kind = 0))
         OR (EXISTS (SELECT 1 FROM audit_events_group_streaming_event_type_filters f
                     WHERE f.external_streaming_destination_id = d.id AND f.kind = 0
                       AND f.audit_event_type = '<event_type>')))

-- After: same as before, plus
    AND (NOT EXISTS (SELECT 1 FROM audit_events_group_streaming_event_type_filters f
                     WHERE f.external_streaming_destination_id = d.id AND f.kind = 1
                       AND f.audit_event_type = '<event_type>'))
DB Lab

Link to DB Lab - https://console.postgres.ai/gitlab/projects/gitlab-production-main/sessions/64704/commands/169689

SELECT 1 AS one
FROM "audit_events_group_external_streaming_destinations"
WHERE "audit_events_group_external_streaming_destinations"."group_id" = 9970
  AND "audit_events_group_external_streaming_destinations"."active" = TRUE
  AND (
    (NOT EXISTS (
      SELECT 1 FROM "audit_events_group_streaming_event_type_filters"
      WHERE "audit_events_group_streaming_event_type_filters"."external_streaming_destination_id" = "audit_events_group_external_streaming_destinations"."id"
        AND "audit_events_group_streaming_event_type_filters"."kind" = 0
    ))
    OR (EXISTS (
      SELECT 1 FROM "audit_events_group_streaming_event_type_filters"
      WHERE "audit_events_group_streaming_event_type_filters"."external_streaming_destination_id" = "audit_events_group_external_streaming_destinations"."id"
        AND "audit_events_group_streaming_event_type_filters"."kind" = 0
        AND "audit_events_group_streaming_event_type_filters"."audit_event_type" = 'repository_git_operation'
    ))
  )
  AND (NOT EXISTS (
    SELECT 1 FROM "audit_events_group_streaming_event_type_filters"
    WHERE "audit_events_group_streaming_event_type_filters"."external_streaming_destination_id" = "audit_events_group_external_streaming_destinations"."id"
      AND "audit_events_group_streaming_event_type_filters"."kind" = 1
      AND "audit_events_group_streaming_event_type_filters"."audit_event_type" = 'repository_git_operation'
  ))
LIMIT 1;
 Limit  (cost=0.69..14.34 rows=1 width=4) (actual time=7.173..7.175 rows=0 loops=1)
   Buffers: shared hit=8 read=7
   I/O Timings: read=7.077 write=0.000
   ->  Nested Loop Anti Join  (cost=0.69..14.34 rows=1 width=4) (actual time=7.172..7.174 rows=0 loops=1)
         Buffers: shared hit=8 read=7
         I/O Timings: read=7.077 write=0.000
         ->  Index Scan using unique_idx_group_destinations_on_name_category_group on public.audit_events_group_external_streaming_destinations  (cost=0.28..7.88 rows=1 width=8) (actual time=7.150..7.152 rows=1 loops=1)
               Index Cond: (audit_events_group_external_streaming_destinations.group_id = 9970)
               Filter: (audit_events_group_external_streaming_destinations.active AND ((NOT EXISTS(SubPlan 1)) OR EXISTS(SubPlan 3)))
               Buffers: shared hit=4 read=7
               I/O Timings: read=7.077 write=0.000
               SubPlan 1
                 ->  Index Scan using uniq_audit_group_event_filters_destination_id_and_event_type on public.audit_events_group_streaming_event_type_filters audit_events_group_streaming_event_type_filters_1  (cost=0.41..207.86 rows=281 width=0) (actual time=4.183..4.183 rows=0 loops=1)
                       Index Cond: (audit_events_group_streaming_event_type_filters_1.external_streaming_destination_id = audit_events_group_external_streaming_destinations.id)
                       Filter: (audit_events_group_streaming_event_type_filters_1.kind = 0)
                       Rows Removed by Filter: 2
                       Buffers: shared hit=1 read=4
                       I/O Timings: read=4.155 write=0.000
               SubPlan 3
                 ->  Index Scan using uniq_audit_group_event_filters_destination_id_and_event_type on public.audit_events_group_streaming_event_type_filters audit_events_group_streaming_event_type_filters_2  (cost=0.41..3.44 rows=1 width=0) (never executed)
                       Index Cond: ((audit_events_group_streaming_event_type_filters_2.external_streaming_destination_id = audit_events_group_external_streaming_destinations.id) AND (audit_events_group_streaming_event_type_filters_2.audit_event_type = 'repository_git_operation'::text))
                       Filter: (audit_events_group_streaming_event_type_filters_2.kind = 0)
                       I/O Timings: read=0.000 write=0.000
         ->  Index Scan using uniq_audit_group_event_filters_destination_id_and_event_type on public.audit_events_group_streaming_event_type_filters  (cost=0.41..3.44 rows=1 width=8) (actual time=0.018..0.018 rows=1 loops=1)
               Index Cond: ((audit_events_group_streaming_event_type_filters.external_streaming_destination_id = audit_events_group_external_streaming_destinations.id) AND (audit_events_group_streaming_event_type_filters.audit_event_type = 'repository_git_operation'::text))
               Filter: (audit_events_group_streaming_event_type_filters.kind = 1)
               Buffers: shared hit=4
               I/O Timings: read=0.000 write=0.000
Settings: work_mem = '230MB', jit = 'off', random_page_cost = '1.5', seq_page_cost = '4', effective_cache_size = '472585MB'
Query ID: -4425077213036035393
Time: 8.627 ms
  - planning: 1.388 ms
  - execution: 7.239 ms
    - I/O read: 7.077 ms
    - I/O write: 0.000 ms

Shared buffers:
  - hits: 8 (~64.00 KiB) from the buffer pool
  - reads: 7 (~56.00 KiB) from the OS file cache, including disk I/O
  - dirtied: 0
  - writes: 0

Link to DB Lab - https://console.postgres.ai/gitlab/projects/gitlab-production-main/sessions/64704/commands/169689

References

Screenshots or screen recordings

Not applicable, no UI changes.

How to set up and validate locally

  1. In a Rails console, create a group destination with a deny filter for repository_git_operation:

    group = Group.where(parent_id: nil).first
dest = AuditEvents::Group::ExternalStreamingDestination.create!(
  group: group, name: 'deny-test', category: :http,
  config: { 'url' => 'https://example.test/deny-test' }
)
AuditEvents::Group::EventTypeFilter.create!(
  external_streaming_destination: dest, namespace: group,
  audit_event_type: 'repository_git_operation', kind: :deny
)

  2. Check the enqueue gate:

    event = AuditEvents::GroupAuditEvent.new(group_id: group.id)
AuditEvents::Streaming::Group::Streamer.new('repository_git_operation', event).streamable?
# => false (was true before this MR)
AuditEvents::Streaming::Group::Streamer.new('user_created', event).streamable?
# => true

Validation after deployment

After this is deployed, the query in https://log.gprd.gitlab.net/app/r/s/3khAS should return zero results.

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.

Edited by Huzaifa Iftikhar

Merge request reports