diff --git a/app/models/hooks/project_hook.rb b/app/models/hooks/project_hook.rb
index 83858ee0c771431de7d4eb83e820129b66517b22..7566b7d23e194077d0c6dd0bc8617825a1e7cdf3 100644
--- a/app/models/hooks/project_hook.rb
+++ b/app/models/hooks/project_hook.rb
@@ -4,6 +4,7 @@ class ProjectHook < WebHook
   include TriggerableHooks
   include Presentable
   include Limitable
+  extend ::Gitlab::Utils::Override
 
   self.limit_scope = :project
 
@@ -33,6 +34,11 @@ def pluralized_name
   def web_hooks_disable_failed?
     Feature.enabled?(:web_hooks_disable_failed, project)
   end
+
+  override :rate_limit
+  def rate_limit
+    project.actual_limits.limit_for(:web_hook_calls)
+  end
 end
 
 ProjectHook.prepend_if_ee('EE::ProjectHook')
diff --git a/app/models/hooks/web_hook.rb b/app/models/hooks/web_hook.rb
index 445381343186ea9cd0b3e92164b2e480d5947912..02b4feb4ccc277e706c86cf0a68abcad7eaf2935 100644
--- a/app/models/hooks/web_hook.rb
+++ b/app/models/hooks/web_hook.rb
@@ -75,6 +75,11 @@ def enable!
     update!(recent_failures: 0, disabled_until: nil, backoff_count: 0)
   end
 
+  # Overridden in ProjectHook and GroupHook, other webhooks are not rate-limited.
+  def rate_limit
+    nil
+  end
+
   private
 
   def web_hooks_disable_failed?
diff --git a/app/services/web_hook_service.rb b/app/services/web_hook_service.rb
index 0535bc625ac1fcbf12ec3aca47282f2860193751..645767200e07d6920e6279b086f98ff000a3f3a4 100644
--- a/app/services/web_hook_service.rb
+++ b/app/services/web_hook_service.rb
@@ -90,7 +90,11 @@ def execute
   end
 
   def async_execute
-    WebHookWorker.perform_async(hook.id, data, hook_name)
+    if rate_limited?(hook)
+      log_rate_limit(hook)
+    else
+      WebHookWorker.perform_async(hook.id, data, hook_name)
+    end
   end
 
   private
@@ -169,4 +173,34 @@ def safe_response_body(response)
 
     response.body.encode('UTF-8', invalid: :replace, undef: :replace, replace: '')
   end
+
+  def rate_limited?(hook)
+    return false unless Feature.enabled?(:web_hooks_rate_limit, default_enabled: :yaml)
+    return false if rate_limit.nil?
+
+    Gitlab::ApplicationRateLimiter.throttled?(
+      :web_hook_calls,
+      scope: [hook],
+      threshold: rate_limit
+    )
+  end
+
+  def rate_limit
+    @rate_limit ||= hook.rate_limit
+  end
+
+  def log_rate_limit(hook)
+    payload = {
+      message: 'Webhook rate limit exceeded',
+      hook_id: hook.id,
+      hook_type: hook.type,
+      hook_name: hook_name
+    }
+
+    Gitlab::AuthLogger.error(payload)
+
+    # Also log into application log for now, so we can use this information
+    # to determine suitable limits for gitlab.com
+    Gitlab::AppLogger.error(payload)
+  end
 end
diff --git a/changelogs/unreleased/329207-rate-limit-webhooks.yml b/changelogs/unreleased/329207-rate-limit-webhooks.yml
new file mode 100644
index 0000000000000000000000000000000000000000..134cdc67c35fd06adfe7192da61f1d1d4e9ee09c
--- /dev/null
+++ b/changelogs/unreleased/329207-rate-limit-webhooks.yml
@@ -0,0 +1,5 @@
+---
+title: Apply rate-limiting to webhook executions
+merge_request: 61151
+author:
+type: performance
diff --git a/config/feature_flags/development/web_hooks_rate_limit.yml b/config/feature_flags/development/web_hooks_rate_limit.yml
new file mode 100644
index 0000000000000000000000000000000000000000..193d51bb250513b7978924611e3583af44921320
--- /dev/null
+++ b/config/feature_flags/development/web_hooks_rate_limit.yml
@@ -0,0 +1,8 @@
+---
+name: web_hooks_rate_limit
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/61151
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/330133
+milestone: '13.12'
+type: development
+group: group::ecosystem
+default_enabled: false
diff --git a/db/migrate/20210503131747_add_web_hook_calls_to_plan_limits.rb b/db/migrate/20210503131747_add_web_hook_calls_to_plan_limits.rb
new file mode 100644
index 0000000000000000000000000000000000000000..cb23c9391eaf06f58aaf42c122058b61f33fcfc2
--- /dev/null
+++ b/db/migrate/20210503131747_add_web_hook_calls_to_plan_limits.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddWebHookCallsToPlanLimits < ActiveRecord::Migration[6.0]
+  def change
+    add_column :plan_limits, :web_hook_calls, :integer, null: false, default: 0
+  end
+end
diff --git a/db/schema_migrations/20210503131747 b/db/schema_migrations/20210503131747
new file mode 100644
index 0000000000000000000000000000000000000000..52771dcb5ded2a152c15154c440d587461018c51
--- /dev/null
+++ b/db/schema_migrations/20210503131747
@@ -0,0 +1 @@
+583c350d82c4d02e910f2c16ed2ec55ccdc880c87b55bf7bd6be3e1839958732
\ No newline at end of file
diff --git a/db/structure.sql b/db/structure.sql
index 976ac394aaeb912961f5106313bb99464177e6fd..15aad6803cb898cd7d39beaba511980191aa91e5 100644
--- a/db/structure.sql
+++ b/db/structure.sql
@@ -16038,7 +16038,8 @@ CREATE TABLE plan_limits (
     terraform_module_max_file_size bigint DEFAULT 1073741824 NOT NULL,
     helm_max_file_size bigint DEFAULT 5242880 NOT NULL,
     ci_registered_group_runners integer DEFAULT 1000 NOT NULL,
-    ci_registered_project_runners integer DEFAULT 1000 NOT NULL
+    ci_registered_project_runners integer DEFAULT 1000 NOT NULL,
+    web_hook_calls integer DEFAULT 0 NOT NULL
 );
 
 CREATE SEQUENCE plan_limits_id_seq
diff --git a/doc/administration/instance_limits.md b/doc/administration/instance_limits.md
index 044a7403e8bda1b4e26d2e557b657df304121348..9ea76ff615135be6f9f36567f19539e015097a7f 100644
--- a/doc/administration/instance_limits.md
+++ b/doc/administration/instance_limits.md
@@ -112,6 +112,49 @@ Limit the maximum daily member invitations allowed per group hierarchy.
 - GitLab.com: Free members may invite 20 members per day.
 - Self-managed: Invites are not limited.
 
+### Webhook calls
+
+> - [Introduced](https://gitlab.com/gitlab-org/gitlab/-/merge_requests/61151) in GitLab 13.12.
+> - [Deployed behind a feature flag](../user/feature_flags.md), disabled by default.
+> - Disabled on GitLab.com.
+> - Not recommended for production use.
+> - To use in GitLab self-managed instances, ask a GitLab administrator to [enable it](#enable-or-disable-rate-limiting-for-webhooks). **(FREE SELF)**
+
+Limit the number of times any given webhook can be called per minute.
+This only applies to project and group webhooks.
+
+Calls over the rate limit are logged into `auth.log`.
+
+```ruby
+# If limits don't exist for the default plan, you can create one with:
+# Plan.default.create_limits!
+
+Plan.default.actual_limits.update!(web_hook_calls: 10)
+```
+
+Set the limit to `0` to disable it.
+
+- **Default rate limit**: Disabled.
+
+#### Enable or disable rate limiting for webhooks **(FREE SELF)**
+
+Rate limiting for webhooks is under development and not ready for production use. It is
+deployed behind a feature flag that is **disabled by default**.
+[GitLab administrators with access to the GitLab Rails console](../administration/feature_flags.md)
+can enable it.
+
+To enable it:
+
+```ruby
+Feature.enable(:web_hooks_rate_limit)
+```
+
+To disable it:
+
+```ruby
+Feature.disable(:web_hooks_rate_limit)
+```
+
 ## Gitaly concurrency limit
 
 Clone traffic can put a large strain on your Gitaly service. To prevent such workloads from overwhelming your Gitaly server, you can set concurrency limits in Gitaly's configuration file.
diff --git a/ee/app/models/hooks/group_hook.rb b/ee/app/models/hooks/group_hook.rb
index f87c537e4272231c2954415cfa2e19e849a19e9f..5f0d4b0d050800aacedf8d86b819fc282ca659a6 100644
--- a/ee/app/models/hooks/group_hook.rb
+++ b/ee/app/models/hooks/group_hook.rb
@@ -4,6 +4,7 @@ class GroupHook < WebHook
   include CustomModelNaming
   include TriggerableHooks
   include Limitable
+  extend ::Gitlab::Utils::Override
 
   self.limit_name = 'group_hooks'
   self.limit_scope = :group
@@ -36,4 +37,9 @@ def pluralized_name
   def web_hooks_disable_failed?
     Feature.enabled?(:web_hooks_disable_failed, group)
   end
+
+  override :rate_limit
+  def rate_limit
+    group.actual_limits.limit_for(:web_hook_calls)
+  end
 end
diff --git a/ee/spec/factories/group_hooks.rb b/ee/spec/factories/group_hooks.rb
index 94558e3218bca56411a9c6819c264beb6c42a002..28360349179d0915050eebfa03a9f01deaa91371 100644
--- a/ee/spec/factories/group_hooks.rb
+++ b/ee/spec/factories/group_hooks.rb
@@ -3,6 +3,7 @@
 FactoryBot.define do
   factory :group_hook do
     url { generate(:url) }
+    group
 
     trait :all_events_enabled do
       push_events { true }
diff --git a/ee/spec/models/hooks/ee/project_hook_spec.rb b/ee/spec/models/hooks/ee/project_hook_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..a05ba710e6cfa4a2969ac9df85d7d9b45a90e8f0
--- /dev/null
+++ b/ee/spec/models/hooks/ee/project_hook_spec.rb
@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe EE::ProjectHook do
+  describe '#rate_limit' do
+    let_it_be(:default_limits) { create(:plan_limits, :default_plan, web_hook_calls: 100) }
+    let_it_be(:ultimate_limits) { create(:plan_limits, plan: create(:ultimate_plan), web_hook_calls: 500) }
+
+    let_it_be(:group) { create(:group) }
+    let_it_be(:group_ultimate) { create(:group_with_plan, plan: :ultimate_plan) }
+    let_it_be(:project) { create(:project, group: group) }
+    let_it_be(:project_ultimate) { create(:project, group: group_ultimate) }
+
+    let_it_be(:hook) { create(:project_hook, project: project) }
+    let_it_be(:hook_ultimate) { create(:project_hook, project: project_ultimate) }
+
+    it 'returns the default limit for a project without a plan' do
+      expect(hook.rate_limit).to be(100)
+    end
+
+    it 'returns the configured limit for a project with the Ultimate plan' do
+      expect(hook_ultimate.rate_limit).to be(500)
+    end
+  end
+end
diff --git a/ee/spec/models/hooks/group_hook_spec.rb b/ee/spec/models/hooks/group_hook_spec.rb
index 692385a3aa2129fd2d44fe460c401edcf8fc9d51..374b526407a43a2f735bdd66cb5b0c8690059944 100644
--- a/ee/spec/models/hooks/group_hook_spec.rb
+++ b/ee/spec/models/hooks/group_hook_spec.rb
@@ -10,4 +10,23 @@
   it_behaves_like 'includes Limitable concern' do
     subject { build(:group_hook, group: create(:group)) }
   end
+
+  describe '#rate_limit' do
+    let_it_be(:default_limits) { create(:plan_limits, :default_plan, web_hook_calls: 100) }
+    let_it_be(:ultimate_limits) { create(:plan_limits, plan: create(:ultimate_plan), web_hook_calls: 500) }
+
+    let_it_be(:group) { create(:group) }
+    let_it_be(:group_ultimate) { create(:group_with_plan, plan: :ultimate_plan) }
+
+    let_it_be(:hook) { create(:group_hook, group: group) }
+    let_it_be(:hook_ultimate) { create(:group_hook, group: group_ultimate) }
+
+    it 'returns the default limit for a group without a plan' do
+      expect(hook.rate_limit).to be(100)
+    end
+
+    it 'returns the configured limit for a group with the Ultimate plan' do
+      expect(hook_ultimate.rate_limit).to be(500)
+    end
+  end
 end
diff --git a/lib/gitlab/application_rate_limiter.rb b/lib/gitlab/application_rate_limiter.rb
index f74edf2b767d09504af8e42bc5d440511cb3638f..f91a56a0cd2020b08f26e6bc37984ec028f418f0 100644
--- a/lib/gitlab/application_rate_limiter.rb
+++ b/lib/gitlab/application_rate_limiter.rb
@@ -34,6 +34,7 @@ def rate_limits
           group_import:                 { threshold: -> { application_settings.group_import_limit }, interval: 1.minute },
           group_testing_hook:           { threshold: 5, interval: 1.minute },
           profile_add_new_email:        { threshold: 5, interval: 1.minute },
+          web_hook_calls:               { interval: 1.minute },
           profile_resend_email_confirmation:  { threshold: 5, interval: 1.minute },
           update_environment_canary_ingress:  { threshold: 1, interval: 1.minute },
           auto_rollback_deployment:           { threshold: 1, interval: 3.minutes }
diff --git a/spec/models/hooks/project_hook_spec.rb b/spec/models/hooks/project_hook_spec.rb
index 69fbc4c3b4fe4dbd33b025db2f41e9903ae92406..881494652321394112c0d75a10d8b919ddcd0379 100644
--- a/spec/models/hooks/project_hook_spec.rb
+++ b/spec/models/hooks/project_hook_spec.rb
@@ -30,4 +30,13 @@
       expect(described_class.tag_push_hooks).to eq([hook])
     end
   end
+
+  describe '#rate_limit' do
+    let_it_be(:hook) { create(:project_hook) }
+    let_it_be(:plan_limits) { create(:plan_limits, :default_plan, web_hook_calls: 100) }
+
+    it 'returns the default limit' do
+      expect(hook.rate_limit).to be(100)
+    end
+  end
 end
diff --git a/spec/models/hooks/service_hook_spec.rb b/spec/models/hooks/service_hook_spec.rb
index f7045d7ac5e92d73c92787f4e640caffd5cb0fbb..60c0eb265ce8a46f252a15d566d5996ef18fb31f 100644
--- a/spec/models/hooks/service_hook_spec.rb
+++ b/spec/models/hooks/service_hook_spec.rb
@@ -22,4 +22,12 @@
       hook.execute(data)
     end
   end
+
+  describe '#rate_limit' do
+    let(:hook) { build(:service_hook) }
+
+    it 'returns nil' do
+      expect(hook.rate_limit).to be_nil
+    end
+  end
 end
diff --git a/spec/models/hooks/system_hook_spec.rb b/spec/models/hooks/system_hook_spec.rb
index 02e630cbf2727c28551549409283a18d26ef84cf..a72034f1ac5005e95d6ddd998b90e5ca9180aa51 100644
--- a/spec/models/hooks/system_hook_spec.rb
+++ b/spec/models/hooks/system_hook_spec.rb
@@ -169,4 +169,12 @@
       hook.async_execute(data, hook_name)
     end
   end
+
+  describe '#rate_limit' do
+    let(:hook) { build(:system_hook) }
+
+    it 'returns nil' do
+      expect(hook.rate_limit).to be_nil
+    end
+  end
 end
diff --git a/spec/models/plan_limits_spec.rb b/spec/models/plan_limits_spec.rb
index 4259c8b708b3bd2869715599b0673873abc04d80..b8c723b3847f943e4da5deddb5ca13c4016bb097 100644
--- a/spec/models/plan_limits_spec.rb
+++ b/spec/models/plan_limits_spec.rb
@@ -210,6 +210,7 @@
         ci_active_jobs
         storage_size_limit
         daily_invites
+        web_hook_calls
       ] + disabled_max_artifact_size_columns
     end
 
diff --git a/spec/services/web_hook_service_spec.rb b/spec/services/web_hook_service_spec.rb
index 46dab4fa1715ce2778431bb02a8686a6ad9ddf12..f9c09086cf25b59efb71ca1d165dd70d9f41e616 100644
--- a/spec/services/web_hook_service_spec.rb
+++ b/spec/services/web_hook_service_spec.rb
@@ -5,8 +5,9 @@
 RSpec.describe WebHookService do
   include StubRequests
 
-  let(:project) { create(:project) }
-  let(:project_hook) { create(:project_hook) }
+  let_it_be(:project) { create(:project) }
+  let_it_be_with_reload(:project_hook) { create(:project_hook, project: project) }
+
   let(:headers) do
     {
       'Content-Type' => 'application/json',
@@ -60,12 +61,8 @@
   end
 
   describe '#execute' do
-    before do
-      project.hooks << [project_hook]
-    end
-
     context 'when token is defined' do
-      let(:project_hook) { create(:project_hook, :token) }
+      let_it_be(:project_hook) { create(:project_hook, :token) }
 
       it 'POSTs to the webhook URL' do
         stub_full_request(project_hook.url, method: :post)
@@ -89,8 +86,8 @@
     end
 
     context 'when auth credentials are present' do
-      let(:url) {'https://example.org'}
-      let(:project_hook) { create(:project_hook, url: 'https://demo:demo@example.org/') }
+      let_it_be(:url) {'https://example.org'}
+      let_it_be(:project_hook) { create(:project_hook, url: 'https://demo:demo@example.org/') }
 
       it 'uses the credentials' do
         stub_full_request(url, method: :post)
@@ -104,8 +101,8 @@
     end
 
     context 'when auth credentials are partial present' do
-      let(:url) {'https://example.org'}
-      let(:project_hook) { create(:project_hook, url: 'https://demo@example.org/') }
+      let_it_be(:url) {'https://example.org'}
+      let_it_be(:project_hook) { create(:project_hook, url: 'https://demo@example.org/') }
 
       it 'uses the credentials anyways' do
         stub_full_request(url, method: :post)
@@ -147,7 +144,7 @@
     end
 
     context 'when url is not encoded' do
-      let(:project_hook) { create(:project_hook, url: 'http://server.com/my path/') }
+      let_it_be(:project_hook) { create(:project_hook, url: 'http://server.com/my path/') }
 
       it 'handles exceptions' do
         expect(service_instance.execute).to eq(status: :error, message: 'bad URI(is not URI?): "http://server.com/my path/"')
@@ -321,12 +318,98 @@
   end
 
   describe '#async_execute' do
-    let(:system_hook) { create(:system_hook) }
+    def expect_to_perform_worker(hook)
+      expect(WebHookWorker).to receive(:perform_async).with(hook.id, data, 'push_hooks')
+    end
+
+    def expect_to_rate_limit(hook, threshold:, throttled: false)
+      expect(Gitlab::ApplicationRateLimiter).to receive(:throttled?)
+        .with(:web_hook_calls, scope: [hook], threshold: threshold)
+        .and_return(throttled)
+    end
+
+    context 'when rate limiting is not configured' do
+      it 'queues a worker without tracking the call' do
+        expect(Gitlab::ApplicationRateLimiter).not_to receive(:throttled?)
+        expect_to_perform_worker(project_hook)
+
+        service_instance.async_execute
+      end
+    end
 
-    it 'enqueue WebHookWorker' do
-      expect(WebHookWorker).to receive(:perform_async).with(project_hook.id, data, 'push_hooks')
+    context 'when rate limiting is configured' do
+      let_it_be(:threshold) { 3 }
+      let_it_be(:plan_limits) { create(:plan_limits, :default_plan, web_hook_calls: threshold) }
 
-      described_class.new(project_hook, data, 'push_hooks').async_execute
+      it 'queues a worker and tracks the call' do
+        expect_to_rate_limit(project_hook, threshold: threshold)
+        expect_to_perform_worker(project_hook)
+
+        service_instance.async_execute
+      end
+
+      context 'when the hook is throttled (via mock)' do
+        before do
+          expect_to_rate_limit(project_hook, threshold: threshold, throttled: true)
+        end
+
+        it 'does not queue a worker and logs an error' do
+          expect(WebHookWorker).not_to receive(:perform_async)
+
+          payload = {
+            message: 'Webhook rate limit exceeded',
+            hook_id: project_hook.id,
+            hook_type: 'ProjectHook',
+            hook_name: 'push_hooks'
+          }
+
+          expect(Gitlab::AuthLogger).to receive(:error).with(payload)
+          expect(Gitlab::AppLogger).to receive(:error).with(payload)
+
+          service_instance.async_execute
+        end
+      end
+
+      context 'when the hook is throttled (via Redis)', :clean_gitlab_redis_cache do
+        before do
+          # Set a high interval to avoid intermittent failures in CI
+          allow(Gitlab::ApplicationRateLimiter).to receive(:rate_limits).and_return(
+            web_hook_calls: { interval: 1.day }
+          )
+
+          expect_to_perform_worker(project_hook).exactly(threshold).times
+
+          threshold.times { service_instance.async_execute }
+        end
+
+        it 'stops queueing workers and logs errors' do
+          expect(Gitlab::AuthLogger).to receive(:error).twice
+          expect(Gitlab::AppLogger).to receive(:error).twice
+
+          2.times { service_instance.async_execute }
+        end
+
+        it 'still queues workers for other hooks' do
+          other_hook = create(:project_hook)
+
+          expect_to_perform_worker(other_hook)
+
+          described_class.new(other_hook, data, :push_hooks).async_execute
+        end
+      end
+
+      context 'when the feature flag is disabled' do
+        before do
+          stub_feature_flags(web_hooks_rate_limit: false)
+        end
+
+        it 'queues a worker without tracking the call' do
+          expect(Gitlab::ApplicationRateLimiter).not_to receive(:throttled?)
+          expect_to_perform_worker(project_hook)
+
+          service_instance.async_execute
+        end
+      end
     end
   end
 end
diff --git a/spec/workers/web_hook_worker_spec.rb b/spec/workers/web_hook_worker_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..becc7461f2a75dd9fbdf470bec012d63238fb195
--- /dev/null
+++ b/spec/workers/web_hook_worker_spec.rb
@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+require 'spec_helper'
+
+RSpec.describe WebHookWorker do
+  include AfterNextHelpers
+
+  let_it_be(:project_hook) { create(:project_hook) }
+  let_it_be(:data) { { foo: 'bar' } }
+  let_it_be(:hook_name) { 'push_hooks' }
+
+  describe '#perform' do
+    it 'delegates to WebHookService' do
+      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name).to receive(:execute)
+
+      subject.perform(project_hook.id, data, hook_name)
+    end
+  end
+end