Adding table for instance streaming headers

Added table and model for instance level audit events
streaming headers

Changelog: added

db/docs/instance_audit_events_streaming_headers.yml

+---
+table_name: instance_audit_events_streaming_headers
+classes:
+  - AuditEvents::Streaming::InstanceHeader
+feature_categories:
+  - audit_events
+description: Represents a HTTP header sent with streaming audit events to instance level destinations
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/88063
+milestone: '16.1'
+gitlab_schema: gitlab_main

db/migrate/20230516175449_create_instance_audit_events_streaming_headers.rb

+# frozen_string_literal: true
+
+class CreateInstanceAuditEventsStreamingHeaders < Gitlab::Database::Migration[2.1]
+  INDEX_NAME = 'idx_headers_instance_external_audit_event_destination_id'
+  UNIQ_INDEX_NAME = 'idx_instance_external_audit_event_destination_id_key_uniq'
+
+  def change
+    create_table :instance_audit_events_streaming_headers do |t|
+      t.timestamps_with_timezone null: false
+      t.references :instance_external_audit_event_destination,
+        null: false,
+        index: { name: INDEX_NAME },
+        foreign_key: { to_table: 'audit_events_instance_external_audit_event_destinations', on_delete: :cascade }
+      t.text :key, null: false, limit: 255
+      t.text :value, null: false, limit: 255
+
+      t.index [:instance_external_audit_event_destination_id, :key], unique: true, name: UNIQ_INDEX_NAME
+    end
+  end
+end

db/schema_migrations/20230516175449

+7c9152abad40bce13c5a1dd3931c9a81a892f516bc07ec4fa9414b9ae35496a2
\ No newline at end of file

db/structure.sql

@@ -17081,6 +17081,26 @@ CREATE SEQUENCE insights_id_seq
 ALTER SEQUENCE insights_id_seq OWNED BY insights.id;
+CREATE TABLE instance_audit_events_streaming_headers (
+    id bigint NOT NULL,
+    created_at timestamp with time zone NOT NULL,
+    updated_at timestamp with time zone NOT NULL,
+    instance_external_audit_event_destination_id bigint NOT NULL,
+    key text NOT NULL,
+    value text NOT NULL,
+    CONSTRAINT check_d52adbbabb CHECK ((char_length(value) <= 255)),
+    CONSTRAINT check_e92010d531 CHECK ((char_length(key) <= 255))
+);
+
+CREATE SEQUENCE instance_audit_events_streaming_headers_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1;
+
+ALTER SEQUENCE instance_audit_events_streaming_headers_id_seq OWNED BY instance_audit_events_streaming_headers.id;
+
 CREATE TABLE integrations (
     id integer NOT NULL,
     project_id integer,
@@ -25272,6 +25292,8 @@ ALTER TABLE ONLY index_statuses ALTER COLUMN id SET DEFAULT nextval('index_statu
 ALTER TABLE ONLY insights ALTER COLUMN id SET DEFAULT nextval('insights_id_seq'::regclass);
+ALTER TABLE ONLY instance_audit_events_streaming_headers ALTER COLUMN id SET DEFAULT nextval('instance_audit_events_streaming_headers_id_seq'::regclass);
+
 ALTER TABLE ONLY integrations ALTER COLUMN id SET DEFAULT nextval('integrations_id_seq'::regclass);
 ALTER TABLE ONLY internal_ids ALTER COLUMN id SET DEFAULT nextval('internal_ids_id_seq'::regclass);
@@ -27357,6 +27379,9 @@ ALTER TABLE ONLY index_statuses
 ALTER TABLE ONLY insights
     ADD CONSTRAINT insights_pkey PRIMARY KEY (id);
+ALTER TABLE ONLY instance_audit_events_streaming_headers
+    ADD CONSTRAINT instance_audit_events_streaming_headers_pkey PRIMARY KEY (id);
+
 ALTER TABLE ONLY integrations
     ADD CONSTRAINT integrations_pkey PRIMARY KEY (id);
@@ -29508,12 +29533,16 @@ CREATE UNIQUE INDEX idx_environment_merge_requests_unique_index ON deployment_me
 CREATE UNIQUE INDEX idx_external_audit_event_destination_id_key_uniq ON audit_events_streaming_headers USING btree (key, external_audit_event_destination_id);
+CREATE INDEX idx_headers_instance_external_audit_event_destination_id ON instance_audit_events_streaming_headers USING btree (instance_external_audit_event_destination_id);
+
 CREATE INDEX idx_installable_conan_pkgs_on_project_id_id ON packages_packages USING btree (project_id, id) WHERE ((package_type = 3) AND (status = ANY (ARRAY[0, 1])));
 CREATE INDEX idx_installable_helm_pkgs_on_project_id_id ON packages_packages USING btree (project_id, id);
 CREATE INDEX idx_installable_npm_pkgs_on_project_id_name_version_id ON packages_packages USING btree (project_id, name, version, id) WHERE ((package_type = 2) AND (status = 0));
+CREATE UNIQUE INDEX idx_instance_external_audit_event_destination_id_key_uniq ON instance_audit_events_streaming_headers USING btree (instance_external_audit_event_destination_id, key);
+
 CREATE INDEX idx_issues_on_health_status_not_null ON issues USING btree (health_status) WHERE (health_status IS NOT NULL);
 CREATE INDEX idx_issues_on_project_id_and_created_at_and_id_and_state_id ON issues USING btree (project_id, created_at, id, state_id);
@@ -37205,6 +37234,9 @@ ALTER TABLE ONLY board_user_preferences
 ALTER TABLE ONLY vulnerability_occurrence_pipelines
     ADD CONSTRAINT fk_rails_dc3ae04693 FOREIGN KEY (occurrence_id) REFERENCES vulnerability_occurrences(id) ON DELETE CASCADE;
+ALTER TABLE ONLY instance_audit_events_streaming_headers
+    ADD CONSTRAINT fk_rails_dc933c1f3c FOREIGN KEY (instance_external_audit_event_destination_id) REFERENCES audit_events_instance_external_audit_event_destinations(id) ON DELETE CASCADE;
+
 ALTER TABLE ONLY deployment_merge_requests
     ADD CONSTRAINT fk_rails_dcbce9f4df FOREIGN KEY (deployment_id) REFERENCES deployments(id) ON DELETE CASCADE;

ee/app/models/audit_events/external_audit_event_destination.rb

@@ -5,8 +5,6 @@ class ExternalAuditEventDestination < ApplicationRecord
     include ExternallyDestinationable
     include Limitable
 
-    MAXIMUM_HEADER_COUNT = 20
-
     self.limit_name = 'external_audit_event_destinations'
     self.limit_scope = :group
     self.table_name = 'audit_events_external_audit_event_destinations'
@@ -15,13 +13,8 @@ class ExternalAuditEventDestination < ApplicationRecord
     has_many :headers, class_name: 'AuditEvents::Streaming::Header'
     has_many :event_type_filters, class_name: 'AuditEvents::Streaming::EventTypeFilter'
 
-    validate :has_fewer_than_20_headers?
     validate :root_level_group?
 
-    def headers_hash
-      { STREAMING_TOKEN_HEADER_KEY => verification_token }.merge(headers.map(&:to_hash).inject(:merge).to_h)
-    end
-
     # TODO: Remove audit_operation.present? guard clause once we implement names for all the audit event types.
     # Epic: https://gitlab.com/groups/gitlab-org/-/epics/8497
     def allowed_to_stream?(audit_operation)
@@ -33,12 +26,6 @@ def allowed_to_stream?(audit_operation)
 
     private
 
-    def has_fewer_than_20_headers?
-      return unless headers.count > MAXIMUM_HEADER_COUNT
-
-      errors.add(:headers, "are limited to #{MAXIMUM_HEADER_COUNT} per destination")
-    end
-
     def root_level_group?
       errors.add(:group, 'must not be a subgroup') if group.subgroup?
     end

ee/app/models/audit_events/instance_external_audit_event_destination.rb

@@ -9,6 +9,8 @@ class InstanceExternalAuditEventDestination < ApplicationRecord
     self.limit_scope = Limitable::GLOBAL_SCOPE
     self.table_name = 'audit_events_instance_external_audit_event_destinations'
 
+    has_many :headers, class_name: 'AuditEvents::Streaming::InstanceHeader'
+
     attr_encrypted :verification_token,
       mode: :per_attribute_iv,
       algorithm: 'aes-256-gcm',
@@ -16,10 +18,6 @@ class InstanceExternalAuditEventDestination < ApplicationRecord
       encode: false,
       encode_iv: false
 
-    def headers_hash
-      { STREAMING_TOKEN_HEADER_KEY => verification_token }
-    end
-
     def allowed_to_stream?(*)
       true
     end

ee/app/models/audit_events/streaming/header.rb

@@ -3,19 +3,16 @@
 module AuditEvents
   module Streaming
     class Header < ApplicationRecord
+      include StreamableHeader
+
       self.table_name = 'audit_events_streaming_headers'
 
       validates :key,
         presence: true,
         length: { maximum: 255 },
         uniqueness: { scope: :external_audit_event_destination_id }
-      validates :value, presence: true, length: { maximum: 255 }
 
       belongs_to :external_audit_event_destination
-
-      def to_hash
-        { key => value }
-      end
     end
   end
 end

ee/app/models/audit_events/streaming/instance_header.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Streaming
+    class InstanceHeader < ApplicationRecord
+      include StreamableHeader
+
+      self.table_name = 'instance_audit_events_streaming_headers'
+
+      validates :key,
+        presence: true,
+        length: { maximum: 255 },
+        uniqueness: { scope: :instance_external_audit_event_destination_id }
+
+      belongs_to :instance_external_audit_event_destination
+    end
+  end
+end

ee/app/models/concerns/audit_events/externally_destinationable.rb

@@ -5,18 +5,32 @@ module ExternallyDestinationable
     extend ActiveSupport::Concern
 
     STREAMING_TOKEN_HEADER_KEY = "X-Gitlab-Event-Streaming-Token"
+    MAXIMUM_HEADER_COUNT = 20
 
     included do
       validates :destination_url, public_url: true, presence: true
       validates :destination_url, uniqueness: true, length: { maximum: 255 }
       validates :verification_token, length: { in: 16..24 }, allow_nil: true
       validates :verification_token, presence: true, on: :update
+      validate :no_more_than_20_headers?
 
       has_secure_token :verification_token, length: 24
 
       def audit_details
         destination_url
       end
+
+      def headers_hash
+        { STREAMING_TOKEN_HEADER_KEY => verification_token }.merge(headers.map(&:to_hash).inject(:merge).to_h)
+      end
+
+      private
+
+      def no_more_than_20_headers?
+        return unless headers.count > MAXIMUM_HEADER_COUNT
+
+        errors.add(:headers, "are limited to #{MAXIMUM_HEADER_COUNT} per destination")
+      end
     end
   end
 end

ee/app/models/concerns/audit_events/streaming/streamable_header.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Streaming
+    module StreamableHeader
+      extend ActiveSupport::Concern
+
+      included do
+        validates :value, presence: true, length: { maximum: 255 }
+
+        def to_hash
+          { key => value }
+        end
+      end
+    end
+  end
+end

ee/spec/factories/audit_events/streaming/instance_headers.rb

+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :instance_audit_events_streaming_header, class: 'AuditEvents::Streaming::InstanceHeader' do
+    sequence :key do |i|
+      "key-#{i}"
+    end
+    value { 'bar' }
+    instance_external_audit_event_destination
+  end
+end

ee/spec/models/audit_events/instance_external_audit_event_destination_spec.rb

@@ -13,11 +13,47 @@
 
   it_behaves_like 'includes Limitable concern'
 
+  describe 'Validations' do
+    it { is_expected.to have_many(:headers).class_name('AuditEvents::Streaming::InstanceHeader') }
+
+    it 'can have 20 headers' do
+      create_list(:instance_audit_events_streaming_header, 20, instance_external_audit_event_destination: subject)
+
+      expect(subject).to be_valid
+    end
+
+    it 'can have no more than 20 headers' do
+      create_list(:instance_audit_events_streaming_header, 21, instance_external_audit_event_destination: subject)
+
+      expect(subject).not_to be_valid
+      expect(subject.errors.full_messages).to contain_exactly('Headers are limited to 20 per destination')
+    end
+  end
+
   describe '#headers_hash' do
     subject { destination.headers_hash }
 
-    it do
-      is_expected.to eq({ 'X-Gitlab-Event-Streaming-Token' => destination.verification_token })
+    context "when destination has 2 headers" do
+      before do
+        create(:instance_audit_events_streaming_header, instance_external_audit_event_destination: destination,
+          key: 'X-GitLab-Hello')
+        create(:instance_audit_events_streaming_header, instance_external_audit_event_destination: destination,
+          key: 'X-GitLab-World')
+      end
+
+      it do
+        is_expected.to eq({ 'X-GitLab-Hello' => 'bar',
+                            'X-GitLab-World' => 'bar',
+                            'X-Gitlab-Event-Streaming-Token' => destination.verification_token })
+      end
+    end
+
+    it 'must have a unique destination_url', :aggregate_failures do
+      create(:instance_external_audit_event_destination, destination_url: 'https://example.com/1')
+      dup = build(:instance_external_audit_event_destination, destination_url: 'https://example.com/1')
+
+      expect(dup).to be_invalid
+      expect(dup.errors.full_messages).to include('Destination url has already been taken')
     end
   end
 end

ee/spec/models/audit_events/streaming/header_spec.rb

@@ -6,17 +6,9 @@
   subject(:header) { build(:audit_events_streaming_header, key: 'foo', value: 'bar') }
 
   describe 'Validations' do
-    it { is_expected.to validate_presence_of(:key) }
-    it { is_expected.to validate_presence_of(:value) }
-    it { is_expected.to validate_length_of(:key).is_at_most(255) }
-    it { is_expected.to validate_length_of(:value).is_at_most(255) }
     it { is_expected.to belong_to(:external_audit_event_destination) }
     it { is_expected.to validate_uniqueness_of(:key).scoped_to(:external_audit_event_destination_id) }
   end
 
-  describe '#to_hash' do
-    subject { header.to_hash }
-
-    it { is_expected.to eq({ 'foo' => 'bar' }) }
-  end
+  include_examples 'audit event streaming header'
 end

ee/spec/models/audit_events/streaming/instance_header_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AuditEvents::Streaming::InstanceHeader, feature_category: :audit_events do
+  subject(:header) { build(:instance_audit_events_streaming_header, key: 'foo', value: 'bar') }
+
+  describe 'Validations' do
+    it { is_expected.to belong_to(:instance_external_audit_event_destination) }
+    it { is_expected.to validate_uniqueness_of(:key).scoped_to(:instance_external_audit_event_destination_id) }
+  end
+
+  include_examples 'audit event streaming header'
+end

ee/spec/support/shared_examples/audit_events/streaming_headers_shared_examples.rb

+# frozen_string_literal: true
+
+RSpec.shared_examples 'audit event streaming header' do
+  describe 'validations' do
+    it { is_expected.to validate_presence_of(:key) }
+    it { is_expected.to validate_presence_of(:value) }
+    it { is_expected.to validate_length_of(:key).is_at_most(255) }
+    it { is_expected.to validate_length_of(:value).is_at_most(255) }
+  end
+
+  describe '#to_hash' do
+    it 'returns the correct hash' do
+      expect(subject.to_hash).to eq({ 'foo' => 'bar' })
+    end
+  end
+end