Merge branch 'andrew-jung/dont-double-stream-audit-events-with-ff' into 'master'

Stream audit events only once using feature flag

See merge request https://gitlab.com/gitlab-org/gitlab/-/merge_requests/182587



Merged-by: Hitesh Raghuvanshi <hraghuvanshi@gitlab.com>
Approved-by: Hitesh Raghuvanshi <hraghuvanshi@gitlab.com>
Reviewed-by: Ahmed Hemdan <ahemdan@gitlab.com>
Co-authored-by: Andrew Jung <ajung@gitlab.com>

ee/lib/audit_events/external_destination_streamer.rb

@@ -19,16 +19,19 @@ def initialize(event_name, audit_event)
     end
 
     def stream_to_destinations
-      streamable_strategies.each(&:execute)
-      return unless feature_flag_enabled?
-
-      streamers.each(&:execute)
+      if feature_flag_enabled? && streamers.any?(&:streamable?)
+        streamers.each(&:execute)
+      else
+        streamable_strategies.each(&:execute)
+      end
     end
 
     def streamable?
-      return !streamable_strategies.empty? unless feature_flag_enabled?
-
-      !streamable_strategies.empty? || streamers.any?(&:streamable?)
+      if feature_flag_enabled?
+        streamers.any?(&:streamable?) || streamable_strategies.any?
+      else
+        streamable_strategies.any?
+      end
     end
 
     private

ee/spec/lib/audit_events/external_destination_streamer_spec.rb

@@ -35,12 +35,34 @@
     context 'when no external streaming destinations are present' do
       it_behaves_like 'external destination streamer'
 
+      it 'only uses the streamers with feature flag enabled' do
+        allow(streamer.send(:streamers)).to receive(:any?).and_return(true)
+        allow_next_instance_of(AuditEvents::Streaming::Group::Streamer) do |instance|
+          allow(instance).to receive(:execute)
+        end
+        allow_next_instance_of(AuditEvents::Streaming::Instance::Streamer) do |instance|
+          allow(instance).to receive(:execute)
+        end
+
+        expect(streamer).to receive(:streamers).at_least(:once).and_call_original
+        expect(streamer).not_to receive(:streamable_strategies)
+
+        stream_to_destinations
+      end
+
       context 'with feature flag disabled' do
         before do
           stub_feature_flags(audit_events_external_destination_streamer_consolidation_refactor: false)
         end
 
         it_behaves_like 'external destination streamer'
+
+        it 'only uses the streamable_strategies with feature flag disabled' do
+          expect(streamer).not_to receive(:streamers)
+          expect(streamer).to receive(:streamable_strategies).at_least(:once).and_call_original
+
+          stream_to_destinations
+        end
       end
     end
 
@@ -73,17 +95,26 @@
 
         expect_next_instance_of(AuditEvents::Streaming::Group::Streamer) do |streamer|
           expect(streamer).to receive(:execute).once.and_call_original
-          expect(streamer).to receive(:streamable?).once.and_return(true)
+          allow(streamer).to receive(:streamable?).and_return(true)
         end
 
         expect_next_instance_of(AuditEvents::Streaming::Instance::Streamer) do |streamer|
           expect(streamer).to receive(:execute).once.and_call_original
-          expect(streamer).to receive(:streamable?).once.and_return(true)
+          allow(streamer).to receive(:streamable?).and_return(true)
         end
 
         stream_to_destinations
       end
 
+      it 'only uses the streamers with feature flag enabled' do
+        allow(streamer.send(:streamers)).to receive(:any?).and_return(true)
+
+        expect(streamer).to receive(:streamers).at_least(:once).and_call_original
+        expect(streamer).not_to receive(:streamable_strategies)
+
+        stream_to_destinations
+      end
+
       context 'with feature flag disabled' do
         before do
           stub_feature_flags(audit_events_external_destination_streamer_consolidation_refactor: false)
@@ -103,70 +134,121 @@
 
           stream_to_destinations
         end
+
+        it 'only uses the streamable_strategies with feature flag disabled' do
+          expect(streamer).not_to receive(:streamers)
+          expect(streamer).to receive(:streamable_strategies).at_least(:once).and_call_original
+
+          stream_to_destinations
+        end
       end
     end
+  end
+
+  describe '#streamable?' do
+    subject(:streamable) { streamer.streamable? }
+
+    context 'when none of the external streaming destinations are present' do
+      it { is_expected.to be_falsey }
+
+      it 'only checks streamers with feature flag enabled' do
+        expect(streamer).to receive(:streamers).at_least(:once).and_call_original
 
-    describe '#streamable?' do
-      subject { streamer.streamable? }
+        expect(streamer).to receive(:streamable_strategies).at_least(:once).and_call_original
+
+        streamable
+      end
+
+      context 'with feature flag disabled' do
+        before do
+          stub_feature_flags(audit_events_external_destination_streamer_consolidation_refactor: false)
+        end
 
-      context 'when none of the external streaming destinations are present' do
         it { is_expected.to be_falsey }
 
-        context 'with feature flag disabled' do
-          before do
-            stub_feature_flags(audit_events_external_destination_streamer_consolidation_refactor: false)
-          end
+        it 'only checks streamable_strategies with feature flag disabled' do
+          expect(streamer).not_to receive(:streamers)
+          expect(streamer).to receive(:streamable_strategies).at_least(:once).and_call_original
 
-          it { is_expected.to be_falsey }
+          streamable
         end
       end
+    end
+
+    context 'when external streaming destinations are present' do
+      before do
+        create(:audit_events_group_external_streaming_destination, :http,
+          group: group)
+        create(:audit_events_instance_external_streaming_destination, :gcp)
+      end
 
-      context 'when external streaming destinations are present' do
+      it { is_expected.to be_truthy }
+
+      it 'only checks streamers with feature flag enabled' do
+        allow_next_instance_of(AuditEvents::Streaming::Group::Streamer) do |instance|
+          allow(instance).to receive(:streamable?).and_return(true)
+        end
+
+        expect(streamer).to receive(:streamers).at_least(:once).and_call_original
+        allow(streamer).to receive(:streamable_strategies).and_return([])
+
+        streamable
+      end
+
+      context 'with feature flag disabled' do
         before do
-          create(:audit_events_group_external_streaming_destination, :http,
-            group: group)
-          create(:audit_events_instance_external_streaming_destination, :gcp)
+          stub_feature_flags(audit_events_external_destination_streamer_consolidation_refactor: false)
+          create(:external_audit_event_destination, group: group)
+          create(:instance_external_audit_event_destination)
+          create(:google_cloud_logging_configuration, group: group)
+          create(:instance_google_cloud_logging_configuration)
+          create(:amazon_s3_configuration, group: group)
+          create(:instance_amazon_s3_configuration)
         end
 
         it { is_expected.to be_truthy }
 
-        context 'with feature flag disabled' do
-          before do
-            stub_feature_flags(audit_events_external_destination_streamer_consolidation_refactor: false)
-            create(:external_audit_event_destination, group: group)
-            create(:instance_external_audit_event_destination)
-            create(:google_cloud_logging_configuration, group: group)
-            create(:instance_google_cloud_logging_configuration)
-            create(:amazon_s3_configuration, group: group)
-            create(:instance_amazon_s3_configuration)
-          end
+        it 'only checks streamable_strategies with feature flag disabled' do
+          expect(streamer).not_to receive(:streamers)
+          expect(streamer).to receive(:streamable_strategies).at_least(:once).and_call_original
 
-          it { is_expected.to be_truthy }
+          streamable
         end
       end
+    end
+
+    context 'when only one destination type is present' do
+      using RSpec::Parameterized::TableSyntax
+
+      where(:factory, :trait, :is_group_destination) do
+        :audit_events_group_external_streaming_destination | :aws | true
+        :audit_events_group_external_streaming_destination | :gcp | true
+        :audit_events_group_external_streaming_destination | :http | true
+        :audit_events_instance_external_streaming_destination | :aws | false
+        :audit_events_instance_external_streaming_destination | :gcp | false
+        :audit_events_instance_external_streaming_destination | :http | false
+      end
 
-      context 'when only one destination type is present' do
-        using RSpec::Parameterized::TableSyntax
-
-        where(:factory, :trait, :is_group_destination) do
-          :audit_events_group_external_streaming_destination | :aws | true
-          :audit_events_group_external_streaming_destination | :gcp | true
-          :audit_events_group_external_streaming_destination | :http | true
-          :audit_events_instance_external_streaming_destination | :aws | false
-          :audit_events_instance_external_streaming_destination | :gcp | false
-          :audit_events_instance_external_streaming_destination | :http | false
-        end
-
-        with_them do
-          before do
-            if is_group_destination
-              create(factory, trait, group: group)
-            else
-              create(factory, trait)
-            end
+      with_them do
+        before do
+          if is_group_destination
+            create(factory, trait, group: group)
+          else
+            create(factory, trait)
           end
+        end
+
+        it { is_expected.to be_truthy }
+
+        it 'only checks streamers with feature flag enabled' do
+          allow_next_instance_of(AuditEvents::Streaming::Group::Streamer) do |instance|
+            allow(instance).to receive(:streamable?).and_return(true)
+          end
+
+          expect(streamer).to receive(:streamers).at_least(:once).and_call_original
+          allow(streamer).to receive(:streamable_strategies).and_return([])
 
-          it { is_expected.to be_truthy }
+          streamable
         end
       end
     end