Use separate Kubernetes namespaces per environment

Kubernetes deployments on new clusters will now have
a separate namespace per project environment, instead
of sharing a single namespace for the project.

Behaviour of existing clusters is unchanged.

All new functionality is controlled by the
:kubernetes_namespace_per_environment feature flag,
which is safe to enable/disable at any time.

42 files changed, 968 insertions, 542 deletions

diff --git a/app/finders/clusters/knative_services_finder.rb b/app/finders/clusters/knative_services_finder.rb
index 7d3b53ef663..71cebe4495e 100644
--- a/app/finders/clusters/knative_services_finder.rb
+++ b/app/finders/clusters/knative_services_finder.rb
@@ -13,11 +13,11 @@ module Clusters
     self.reactive_cache_key = ->(finder) { finder.model_name }
     self.reactive_cache_worker_finder = ->(_id, *cache_args) { from_cache(*cache_args) }
 
-    attr_reader :cluster, :project
+    attr_reader :cluster, :environment
 
-    def initialize(cluster, project)
+    def initialize(cluster, environment)
       @cluster = cluster
-      @project = project
+      @environment = environment
     end
 
     def with_reactive_cache_memoized(*cache_args, &block)
@@ -30,11 +30,11 @@ module Clusters
       clear_reactive_cache!(*cache_args)
     end
 
-    def self.from_cache(cluster_id, project_id)
+    def self.from_cache(cluster_id, environment_id)
       cluster = Clusters::Cluster.find(cluster_id)
-      project = ::Project.find(project_id)
+      environment = Environment.find(environment_id)
 
-      new(cluster, project)
+      new(cluster, environment)
     end
 
     def calculate_reactive_cache(*)
@@ -56,7 +56,7 @@ module Clusters
     end
 
     def cache_args
-      [cluster.id, project.id]
+      [cluster.id, environment.id]
     end
 
     def service_pod_details(service)
@@ -84,7 +84,7 @@ module Clusters
     private
 
     def search_namespace
-      @search_namespace ||= cluster.kubernetes_namespace_for(project)
+      @search_namespace ||= cluster.kubernetes_namespace_for(environment)
     end
 
     def knative_client
diff --git a/app/finders/clusters/kubernetes_namespace_finder.rb b/app/finders/clusters/kubernetes_namespace_finder.rb
new file mode 100644
index 00000000000..e947796c1e7
--- /dev/null
+++ b/app/finders/clusters/kubernetes_namespace_finder.rb
@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+module Clusters
+  class KubernetesNamespaceFinder
+    attr_reader :cluster, :project, :environment_slug
+
+    def initialize(cluster, project:, environment_slug:, allow_blank_token: false)
+      @cluster = cluster
+      @project = project
+      @environment_slug = environment_slug
+      @allow_blank_token = allow_blank_token
+    end
+
+    def execute
+      find_namespace(with_environment: cluster.namespace_per_environment?)
+    end
+
+    private
+
+    attr_reader :allow_blank_token
+
+    def find_namespace(with_environment:)
+      relation = with_environment ? namespaces.with_environment_slug(environment_slug) : namespaces
+
+      relation.find_by_project_id(project.id)
+    end
+
+    def namespaces
+      if allow_blank_token
+        cluster.kubernetes_namespaces
+      else
+        cluster.kubernetes_namespaces.has_service_account_token
+      end
+    end
+  end
+end
diff --git a/app/finders/projects/serverless/functions_finder.rb b/app/finders/projects/serverless/functions_finder.rb
index ebe50806ca1..e8c50ef1a88 100644
--- a/app/finders/projects/serverless/functions_finder.rb
+++ b/app/finders/projects/serverless/functions_finder.rb
@@ -3,10 +3,11 @@
 module Projects
   module Serverless
     class FunctionsFinder
+      include Gitlab::Utils::StrongMemoize
+
       attr_reader :project
 
       def initialize(project)
-        @clusters = project.clusters
         @project = project
       end
 
@@ -16,9 +17,8 @@ module Projects
 
       # Possible return values: Clusters::KnativeServicesFinder::KNATIVE_STATE
       def knative_installed
-        states = @clusters.map do |cluster|
-          cluster.application_knative
-          cluster.knative_services_finder(project).knative_detected.tap do |state|
+        states = services_finders.map do |finder|
+          finder.knative_detected.tap do |state|
             return state if state == ::Clusters::KnativeServicesFinder::KNATIVE_STATES['checking'] # rubocop:disable Cop/AvoidReturnFromBlocks
           end
         end
@@ -31,66 +31,70 @@ module Projects
       end
 
       def invocation_metrics(environment_scope, name)
-        return unless prometheus_adapter&.can_query?
+        environment = finders_for_scope(environment_scope).first&.environment
 
-        cluster = @clusters.find do |c|
-          environment_scope == c.environment_scope
+        if environment.present? && environment.prometheus_adapter&.can_query?
+          func = ::Serverless::Function.new(project, name, environment.deployment_namespace)
+          environment.prometheus_adapter.query(:knative_invocation, func)
         end
-
-        func = ::Serverless::Function.new(project, name, cluster.kubernetes_namespace_for(project))
-        prometheus_adapter.query(:knative_invocation, func)
       end
 
       def has_prometheus?(environment_scope)
-        @clusters.any? do |cluster|
-          environment_scope == cluster.environment_scope && cluster.application_prometheus_available?
+        finders_for_scope(environment_scope).any? do |finder|
+          finder.cluster.application_prometheus_available?
         end
       end
 
       private
 
       def knative_service(environment_scope, name)
-        @clusters.map do |cluster|
-          next if environment_scope != cluster.environment_scope
-
-          services = cluster
-            .knative_services_finder(project)
+        finders_for_scope(environment_scope).map do |finder|
+          services = finder
             .services
             .select { |svc| svc["metadata"]["name"] == name }
 
-          add_metadata(cluster, services).first unless services.nil?
+          add_metadata(finder, services).first unless services.nil?
         end
       end
 
       def knative_services
-        @clusters.map do |cluster|
-          services = cluster
-            .knative_services_finder(project)
-            .services
+        services_finders.map do |finder|
+          services = finder.services
 
-          add_metadata(cluster, services) unless services.nil?
+          add_metadata(finder, services) unless services.nil?
         end
       end
 
-      def add_metadata(cluster, services)
+      def add_metadata(finder, services)
+        add_pod_count = services.one?
+
         services.each do |s|
-          s["environment_scope"] = cluster.environment_scope
-          s["cluster_id"] = cluster.id
+          s["environment_scope"] = finder.cluster.environment_scope
+          s["cluster_id"] = finder.cluster.id
 
-          if services.length == 1
-            s["podcount"] = cluster
-              .knative_services_finder(project)
+          if add_pod_count
+            s["podcount"] = finder
               .service_pod_details(s["metadata"]["name"])
               .length
           end
         end
       end
 
-      # rubocop: disable CodeReuse/ServiceClass
-      def prometheus_adapter
-        @prometheus_adapter ||= ::Prometheus::AdapterService.new(project).prometheus_adapter
+      def services_finders
+        strong_memoize(:services_finders) do
+          available_environments.map(&:knative_services_finder).compact
+        end
+      end
+
+      def available_environments
+        @project.environments.available.preload_cluster
+      end
+
+      def finders_for_scope(environment_scope)
+        services_finders.select do |finder|
+          environment_scope == finder.cluster.environment_scope
+        end
       end
-      # rubocop: enable CodeReuse/ServiceClass
     end
   end
 end
diff --git a/app/models/clusters/cluster.rb b/app/models/clusters/cluster.rb
index 8bb44b0ce40..97d39491b73 100644
--- a/app/models/clusters/cluster.rb
+++ b/app/models/clusters/cluster.rb
@@ -53,6 +53,7 @@ module Clusters
     validates :name, cluster_name: true
     validates :cluster_type, presence: true
     validates :domain, allow_blank: true, hostname: { allow_numeric_hostname: true }
+    validates :namespace_per_environment, inclusion: { in: [true, false] }
 
     validate :restrict_modification, on: :update
     validate :no_groups, unless: :group_type?
@@ -100,16 +101,6 @@ module Clusters
 
     scope :default_environment, -> { where(environment_scope: DEFAULT_ENVIRONMENT) }
 
-    scope :with_knative_installed, -> { joins(:application_knative).merge(Clusters::Applications::Knative.available) }
-
-    scope :preload_knative, -> {
-      preload(
-        :kubernetes_namespaces,
-        :platform_kubernetes,
-        :application_knative
-      )
-    }
-
     def self.ancestor_clusters_for_clusterable(clusterable, hierarchy_order: :asc)
       return [] if clusterable.is_a?(Instance)
 
@@ -177,36 +168,15 @@ module Clusters
       platform_kubernetes.kubeclient if kubernetes?
     end
 
-    ##
-    # This is subtly different to #find_or_initialize_kubernetes_namespace_for_project
-    # below because it will ignore any namespaces that have not got a service account
-    # token. This provides a guarantee that any namespace selected here can be used
-    # for cluster operations - a namespace needs to have a service account configured
-    # before it it can be used.
-    #
-    # This is used for selecting a namespace to use when querying a cluster, or
-    # generating variables to pass to CI.
-    def kubernetes_namespace_for(project)
-      find_or_initialize_kubernetes_namespace_for_project(
-        project, scope: kubernetes_namespaces.has_service_account_token
-      ).namespace
-    end
-
-    ##
-    # This is subtly different to #kubernetes_namespace_for because it will include
-    # namespaces that have yet to receive a service account token. This allows
-    # the namespace configuration process to be repeatable - if a namespace has
-    # already been created without a token we don't need to create another
-    # record entirely, just set the token on the pre-existing namespace.
-    #
-    # This is used for configuring cluster namespaces.
-    def find_or_initialize_kubernetes_namespace_for_project(project, scope: kubernetes_namespaces)
-      attributes = { project: project }
-      attributes[:cluster_project] = cluster_project if project_type?
+    def kubernetes_namespace_for(environment)
+      project = environment.project
+      persisted_namespace = Clusters::KubernetesNamespaceFinder.new(
+        self,
+        project: project,
+        environment_slug: environment.slug
+      ).execute
 
-      scope.find_or_initialize_by(attributes).tap do |namespace|
-        namespace.set_defaults
-      end
+      persisted_namespace&.namespace || Gitlab::Kubernetes::DefaultNamespace.new(self, project: project).from_environment_slug(environment.slug)
     end
 
     def allow_user_defined_namespace?
@@ -225,10 +195,6 @@ module Clusters
       end
     end
 
-    def knative_services_finder(project)
-      @knative_services_finder ||= KnativeServicesFinder.new(self, project)
-    end
-
     private
 
     def instance_domain
diff --git a/app/models/clusters/kubernetes_namespace.rb b/app/models/clusters/kubernetes_namespace.rb
index b0c4900546e..69a2b99fcb6 100644
--- a/app/models/clusters/kubernetes_namespace.rb
+++ b/app/models/clusters/kubernetes_namespace.rb
@@ -9,12 +9,12 @@ module Clusters
     belongs_to :cluster_project, class_name: 'Clusters::Project'
     belongs_to :cluster, class_name: 'Clusters::Cluster'
     belongs_to :project, class_name: '::Project'
+    belongs_to :environment, optional: true
     has_one :platform_kubernetes, through: :cluster
 
-    before_validation :set_defaults
-
     validates :namespace, presence: true
     validates :namespace, uniqueness: { scope: :cluster_id }
+    validates :environment_id, uniqueness: { scope: [:cluster_id, :project_id] }, allow_nil: true
 
     validates :service_account_name, presence: true
 
@@ -27,6 +27,7 @@ module Clusters
         algorithm: 'aes-256-cbc'
 
     scope :has_service_account_token, -> { where.not(encrypted_service_account_token: nil) }
+    scope :with_environment_slug, -> (slug) { joins(:environment).where(environments: { slug: slug }) }
 
     def token_name
       "#{namespace}-token"
@@ -42,34 +43,8 @@ module Clusters
       end
     end
 
-    def set_defaults
-      self.namespace ||= default_platform_kubernetes_namespace
-      self.namespace ||= default_project_namespace
-      self.service_account_name ||= default_service_account_name
-    end
-
     private
 
-    def default_service_account_name
-      return unless namespace
-
-      "#{namespace}-service-account"
-    end
-
-    def default_platform_kubernetes_namespace
-      platform_kubernetes&.namespace.presence
-    end
-
-    def default_project_namespace
-      Gitlab::NamespaceSanitizer.sanitize(project_slug) if project_slug
-    end
-
-    def project_slug
-      return unless project
-
-      "#{project.path}-#{project.id}".downcase
-    end
-
     def kubeconfig
       to_kubeconfig(
         url: api_url,
diff --git a/app/models/clusters/platforms/kubernetes.rb b/app/models/clusters/platforms/kubernetes.rb
index 9296c28776b..37614fbe3ca 100644
--- a/app/models/clusters/platforms/kubernetes.rb
+++ b/app/models/clusters/platforms/kubernetes.rb
@@ -51,11 +51,6 @@ module Clusters
       delegate :provided_by_user?, to: :cluster, allow_nil: true
       delegate :allow_user_defined_namespace?, to: :cluster, allow_nil: true
 
-      # This is just to maintain compatibility with KubernetesService, which
-      # will be removed in https://gitlab.com/gitlab-org/gitlab-ce/issues/39217.
-      # It can be removed once KubernetesService is gone.
-      delegate :kubernetes_namespace_for, to: :cluster, allow_nil: true
-
       alias_method :active?, :enabled?
 
       enum_with_nil authorization_type: {
@@ -66,7 +61,7 @@ module Clusters
 
       default_value_for :authorization_type, :rbac
 
-      def predefined_variables(project:)
+      def predefined_variables(project:, environment_name:)
         Gitlab::Ci::Variables::Collection.new.tap do |variables|
           variables.append(key: 'KUBE_URL', value: api_url)
 
@@ -77,15 +72,14 @@ module Clusters
           end
 
           if !cluster.managed?
-            project_namespace = namespace.presence || "#{project.path}-#{project.id}".downcase
+            namespace = Gitlab::Kubernetes::DefaultNamespace.new(cluster, project: project).from_environment_name(environment_name)
 
             variables
-              .append(key: 'KUBE_URL', value: api_url)
               .append(key: 'KUBE_TOKEN', value: token, public: false, masked: true)
-              .append(key: 'KUBE_NAMESPACE', value: project_namespace)
-              .append(key: 'KUBECONFIG', value: kubeconfig(project_namespace), public: false, file: true)
+              .append(key: 'KUBE_NAMESPACE', value: namespace)
+              .append(key: 'KUBECONFIG', value: kubeconfig(namespace), public: false, file: true)
 
-          elsif kubernetes_namespace = cluster.kubernetes_namespaces.has_service_account_token.find_by(project: project)
+          elsif kubernetes_namespace = find_persisted_namespace(project, environment_name: environment_name)
             variables.concat(kubernetes_namespace.predefined_variables)
           end
 
@@ -111,6 +105,22 @@ module Clusters
 
       private
 
+      ##
+      # Environment slug can be predicted given an environment
+      # name, so even if the environment isn't persisted yet we
+      # still know what to look for.
+      def environment_slug(name)
+        Gitlab::Slug::Environment.new(name).generate
+      end
+
+      def find_persisted_namespace(project, environment_name:)
+        Clusters::KubernetesNamespaceFinder.new(
+          cluster,
+          project: project,
+          environment_slug: environment_slug(environment_name)
+        ).execute
+      end
+
       def kubeconfig(namespace)
         to_kubeconfig(
           url: api_url,
diff --git a/app/models/environment.rb b/app/models/environment.rb
index 513427ac2c5..1b53c4b45f9 100644
--- a/app/models/environment.rb
+++ b/app/models/environment.rb
@@ -48,6 +48,7 @@ class Environment < ApplicationRecord
   end
   scope :in_review_folder, -> { where(environment_type: "review") }
   scope :for_name, -> (name) { where(name: name) }
+  scope :preload_cluster, -> { preload(last_deployment: :cluster) }
 
   ##
   # Search environments which have names like the given query.
@@ -170,7 +171,7 @@ class Environment < ApplicationRecord
 
   def deployment_namespace
     strong_memoize(:kubernetes_namespace) do
-      deployment_platform&.kubernetes_namespace_for(project)
+      deployment_platform.cluster.kubernetes_namespace_for(self) if deployment_platform
     end
   end
 
@@ -233,6 +234,12 @@ class Environment < ApplicationRecord
     end
   end
 
+  def knative_services_finder
+    if last_deployment&.cluster
+      Clusters::KnativeServicesFinder.new(last_deployment.cluster, self)
+    end
+  end
+
   private
 
   def generate_slug
diff --git a/app/models/project.rb b/app/models/project.rb
index 44b6e5a532c..960795b73cb 100644
--- a/app/models/project.rb
+++ b/app/models/project.rb
@@ -1855,8 +1855,12 @@ class Project < ApplicationRecord
     end
   end
 
-  def deployment_variables(environment: nil)
-    deployment_platform(environment: environment)&.predefined_variables(project: self) || []
+  def deployment_variables(environment:)
+    platform = deployment_platform(environment: environment)
+
+    return [] unless platform.present?
+
+    platform.predefined_variables(project: self, environment_name: environment)
   end
 
   def auto_devops_variables
diff --git a/app/models/project_services/mock_deployment_service.rb b/app/models/project_services/mock_deployment_service.rb
index 1103cb11e73..6f2b0f7747f 100644
--- a/app/models/project_services/mock_deployment_service.rb
+++ b/app/models/project_services/mock_deployment_service.rb
@@ -24,7 +24,7 @@ class MockDeploymentService < Service
     %w()
   end
 
-  def predefined_variables(project:)
+  def predefined_variables(project:, environment_name:)
     []
   end
 
diff --git a/app/services/clusters/build_kubernetes_namespace_service.rb b/app/services/clusters/build_kubernetes_namespace_service.rb
new file mode 100644
index 00000000000..2574f77bbf9
--- /dev/null
+++ b/app/services/clusters/build_kubernetes_namespace_service.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+module Clusters
+  class BuildKubernetesNamespaceService
+    attr_reader :cluster, :environment
+
+    def initialize(cluster, environment:)
+      @cluster = cluster
+      @environment = environment
+    end
+
+    def execute
+      cluster.kubernetes_namespaces.build(attributes)
+    end
+
+    private
+
+    def attributes
+      attributes = {
+        project: environment.project,
+        namespace: namespace,
+        service_account_name: "#{namespace}-service-account"
+      }
+
+      attributes[:cluster_project] = cluster.cluster_project if cluster.project_type?
+      attributes[:environment] = environment if cluster.namespace_per_environment?
+
+      attributes
+    end
+
+    def namespace
+      Gitlab::Kubernetes::DefaultNamespace.new(cluster, project: environment.project).from_environment_slug(environment.slug)
+    end
+  end
+end
diff --git a/app/services/clusters/create_service.rb b/app/services/clusters/create_service.rb
index 5fb5e15c32d..e5a5b73321a 100644
--- a/app/services/clusters/create_service.rb
+++ b/app/services/clusters/create_service.rb
@@ -11,7 +11,8 @@ module Clusters
     def execute(access_token: nil)
       raise ArgumentError, 'Unknown clusterable provided' unless clusterable
 
-      cluster_params = params.merge(user: current_user).merge(clusterable_params)
+      cluster_params = params.merge(global_params).merge(clusterable_params)
+
       cluster_params[:provider_gcp_attributes].try do |provider|
         provider[:access_token] = access_token
       end
@@ -35,6 +36,10 @@ module Clusters
       @clusterable ||= params.delete(:clusterable)
     end
 
+    def global_params
+      { user: current_user, namespace_per_environment: Feature.enabled?(:kubernetes_namespace_per_environment, default_enabled: true) }
+    end
+
     def clusterable_params
       case clusterable
       when ::Project
diff --git a/app/services/clusters/gcp/kubernetes/create_or_update_namespace_service.rb b/app/services/clusters/gcp/kubernetes/create_or_update_namespace_service.rb
index 806f320381d..c45dac7b273 100644
--- a/app/services/clusters/gcp/kubernetes/create_or_update_namespace_service.rb
+++ b/app/services/clusters/gcp/kubernetes/create_or_update_namespace_service.rb
@@ -11,7 +11,6 @@ module Clusters
         end
 
         def execute
-          configure_kubernetes_namespace
           create_project_service_account
           configure_kubernetes_token
 
@@ -22,10 +21,6 @@ module Clusters
 
         attr_reader :cluster, :kubernetes_namespace, :platform
 
-        def configure_kubernetes_namespace
-          kubernetes_namespace.set_defaults
-        end
-
         def create_project_service_account
           Clusters::Gcp::Kubernetes::CreateOrUpdateServiceAccountService.namespace_creator(
             platform.kubeclient,
diff --git a/changelogs/unreleased/52494-separate-namespace-per-project-environment-slug.yml b/changelogs/unreleased/52494-separate-namespace-per-project-environment-slug.yml
new file mode 100644
index 00000000000..645c92127a3
--- /dev/null
+++ b/changelogs/unreleased/52494-separate-namespace-per-project-environment-slug.yml
@@ -0,0 +1,5 @@
+---
+title: Use separate Kubernetes namespaces per environment
+merge_request: 30711
+author:
+type: added
diff --git a/db/migrate/20190712040400_add_environment_id_to_clusters_kubernetes_namespaces.rb b/db/migrate/20190712040400_add_environment_id_to_clusters_kubernetes_namespaces.rb
new file mode 100644
index 00000000000..5ab5a9ba2f8
--- /dev/null
+++ b/db/migrate/20190712040400_add_environment_id_to_clusters_kubernetes_namespaces.rb
@@ -0,0 +1,10 @@
+# frozen_string_literal: true
+
+class AddEnvironmentIdToClustersKubernetesNamespaces < ActiveRecord::Migration[5.1]
+  DOWNTIME = false
+
+  def change
+    add_reference :clusters_kubernetes_namespaces, :environment,
+      index: true, type: :bigint, foreign_key: { on_delete: :nullify }
+  end
+end
diff --git a/db/migrate/20190712040412_index_clusters_kubernetes_namespaces_on_environment_id.rb b/db/migrate/20190712040412_index_clusters_kubernetes_namespaces_on_environment_id.rb
new file mode 100644
index 00000000000..23082492091
--- /dev/null
+++ b/db/migrate/20190712040412_index_clusters_kubernetes_namespaces_on_environment_id.rb
@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+class IndexClustersKubernetesNamespacesOnEnvironmentId < ActiveRecord::Migration[5.1]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+  INDEX_NAME = 'index_kubernetes_namespaces_on_cluster_project_environment_id'
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index :clusters_kubernetes_namespaces, [:cluster_id, :project_id, :environment_id], unique: true, name: INDEX_NAME
+  end
+
+  def down
+    remove_concurrent_index :clusters_kubernetes_namespaces, name: INDEX_NAME
+  end
+end
diff --git a/db/migrate/20190712064021_add_namespace_per_environment_flag_to_clusters.rb b/db/migrate/20190712064021_add_namespace_per_environment_flag_to_clusters.rb
new file mode 100644
index 00000000000..4c8a0ab3def
--- /dev/null
+++ b/db/migrate/20190712064021_add_namespace_per_environment_flag_to_clusters.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+# See http://doc.gitlab.com/ce/development/migration_style_guide.html
+# for more information on how to write migrations for GitLab.
+
+class AddNamespacePerEnvironmentFlagToClusters < ActiveRecord::Migration[5.1]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up
+    add_column_with_default :clusters, :namespace_per_environment, :boolean, default: false
+  end
+
+  def down
+    remove_column :clusters, :namespace_per_environment
+  end
+end
diff --git a/db/schema.rb b/db/schema.rb
index 8d5da4a07dd..828e36aa96c 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -880,6 +880,7 @@ ActiveRecord::Schema.define(version: 2019_08_02_235445) do
     t.integer "cluster_type", limit: 2, default: 3, null: false
     t.string "domain"
     t.boolean "managed", default: true, null: false
+    t.boolean "namespace_per_environment", default: false, null: false
     t.index ["enabled"], name: "index_clusters_on_enabled"
     t.index ["user_id"], name: "index_clusters_on_user_id"
   end
@@ -984,9 +985,12 @@ ActiveRecord::Schema.define(version: 2019_08_02_235445) do
     t.string "encrypted_service_account_token_iv"
     t.string "namespace", null: false
     t.string "service_account_name"
+    t.bigint "environment_id"
     t.index ["cluster_id", "namespace"], name: "kubernetes_namespaces_cluster_and_namespace", unique: true
+    t.index ["cluster_id", "project_id", "environment_id"], name: "index_kubernetes_namespaces_on_cluster_project_environment_id", unique: true
     t.index ["cluster_id"], name: "index_clusters_kubernetes_namespaces_on_cluster_id"
     t.index ["cluster_project_id"], name: "index_clusters_kubernetes_namespaces_on_cluster_project_id"
+    t.index ["environment_id"], name: "index_clusters_kubernetes_namespaces_on_environment_id"
     t.index ["project_id"], name: "index_clusters_kubernetes_namespaces_on_project_id"
   end
 
@@ -3711,6 +3715,7 @@ ActiveRecord::Schema.define(version: 2019_08_02_235445) do
   add_foreign_key "clusters_applications_runners", "clusters", on_delete: :cascade
   add_foreign_key "clusters_kubernetes_namespaces", "cluster_projects", on_delete: :nullify
   add_foreign_key "clusters_kubernetes_namespaces", "clusters", on_delete: :cascade
+  add_foreign_key "clusters_kubernetes_namespaces", "environments", on_delete: :nullify
   add_foreign_key "clusters_kubernetes_namespaces", "projects", on_delete: :nullify
   add_foreign_key "container_repositories", "projects"
   add_foreign_key "dependency_proxy_blobs", "namespaces", column: "group_id", name: "fk_db58bbc5d7", on_delete: :cascade
diff --git a/doc/user/project/clusters/index.md b/doc/user/project/clusters/index.md
index f0d80dad94f..7dfd0d04637 100644
--- a/doc/user/project/clusters/index.md
+++ b/doc/user/project/clusters/index.md
@@ -384,13 +384,9 @@ NOTE: **Note:**
 [RBAC](#rbac-cluster-resources) is recommended and the GitLab default.
 
 GitLab creates the necessary service accounts and privileges to install and run
-[GitLab managed applications](#installing-applications). When GitLab creates the cluster:
-
-- A `gitlab` service account with `cluster-admin` privileges is created in the `default` namespace
-  to manage the newly created cluster.
-- A project service account with [`edit`
-  privileges](https://kubernetes.io/docs/reference/access-authn-authz/rbac/#user-facing-roles)
-  is created in the GitLab-created project namespace for [deployment jobs](#deployment-variables).
+[GitLab managed applications](#installing-applications). When GitLab creates the cluster,
+a `gitlab` service account with `cluster-admin` privileges is created in the `default` namespace
+to manage the newly created cluster.
 
   NOTE: **Note:**
   Restricted service account for deployment was [introduced](https://gitlab.com/gitlab-org/gitlab-ce/issues/51716) in GitLab 11.5.
@@ -412,32 +408,37 @@ The resources created by GitLab differ depending on the type of cluster.
 
 GitLab creates the following resources for ABAC clusters.
 
-| Name              | Type                 | Details                           | Created when               |
-|:------------------|:---------------------|:----------------------------------|:---------------------------|
-| `gitlab`          | `ServiceAccount`     | `default` namespace               | Creating a new GKE Cluster |
-| `gitlab-token`    | `Secret`             | Token for `gitlab` ServiceAccount | Creating a new GKE Cluster |
-| `tiller`          | `ServiceAccount`     | `gitlab-managed-apps` namespace   | Installing Helm Tiller     |
-| `tiller-admin`    | `ClusterRoleBinding` | `cluster-admin` roleRef           | Installing Helm Tiller     |
-| Project namespace | `ServiceAccount`     | Uses namespace of Project         | Deploying to a cluster     |
-| Project namespace | `Secret`             | Token for project ServiceAccount  | Deploying to a cluster     |
+| Name                  | Type                 | Details                              | Created when               |
+|:----------------------|:---------------------|:-------------------------------------|:---------------------------|
+| `gitlab`              | `ServiceAccount`     | `default` namespace                         | Creating a new GKE Cluster |
+| `gitlab-token`        | `Secret`             | Token for `gitlab` ServiceAccount           | Creating a new GKE Cluster |
+| `tiller`              | `ServiceAccount`     | `gitlab-managed-apps` namespace             | Installing Helm Tiller     |
+| `tiller-admin`        | `ClusterRoleBinding` | `cluster-admin` roleRef                     | Installing Helm Tiller     |
+| Environment namespace | `Namespace`          | Contains all environment-specific resources | Deploying to a cluster     |
+| Environment namespace | `ServiceAccount`     | Uses namespace of environment               | Deploying to a cluster     |
+| Environment namespace | `Secret`             | Token for environment ServiceAccount        | Deploying to a cluster     |
 
 #### RBAC cluster resources
 
 GitLab creates the following resources for RBAC clusters.
 
-| Name              | Type                 | Details                                                                                                    | Created when               |
-|:------------------|:---------------------|:-----------------------------------------------------------------------------------------------------------|:---------------------------|
-| `gitlab`          | `ServiceAccount`     | `default` namespace                                                                                        | Creating a new GKE Cluster |
-| `gitlab-admin`    | `ClusterRoleBinding` | [`cluster-admin`](https://kubernetes.io/docs/reference/access-authn-authz/rbac/#user-facing-roles) roleRef | Creating a new GKE Cluster |
-| `gitlab-token`    | `Secret`             | Token for `gitlab` ServiceAccount                                                                          | Creating a new GKE Cluster |
-| `tiller`          | `ServiceAccount`     | `gitlab-managed-apps` namespace                                                                            | Installing Helm Tiller     |
-| `tiller-admin`    | `ClusterRoleBinding` | `cluster-admin` roleRef                                                                                    | Installing Helm Tiller     |
-| Project namespace | `ServiceAccount`     | Uses namespace of Project                                                                                  | Deploying to a cluster     |
-| Project namespace | `Secret`             | Token for project ServiceAccount                                                                           | Deploying to a cluster     |
-| Project namespace | `RoleBinding`        | [`edit`](https://kubernetes.io/docs/reference/access-authn-authz/rbac/#user-facing-roles) roleRef          | Deploying to a cluster     |
+| Name                  | Type                 | Details                                                                                                    | Created when               |
+|:----------------------|:---------------------|:-----------------------------------------------------------------------------------------------------------|:---------------------------|
+| `gitlab`              | `ServiceAccount`     | `default` namespace                                                                                        | Creating a new GKE Cluster |
+| `gitlab-admin`        | `ClusterRoleBinding` | [`cluster-admin`](https://kubernetes.io/docs/reference/access-authn-authz/rbac/#user-facing-roles) roleRef | Creating a new GKE Cluster |
+| `gitlab-token`        | `Secret`             | Token for `gitlab` ServiceAccount                                                                          | Creating a new GKE Cluster |
+| `tiller`              | `ServiceAccount`     | `gitlab-managed-apps` namespace                                                                            | Installing Helm Tiller     |
+| `tiller-admin`        | `ClusterRoleBinding` | `cluster-admin` roleRef                                                                                    | Installing Helm Tiller     |
+| Environment namespace | `Namespace`          | Contains all environment-specific resources                                                                | Deploying to a cluster     |
+| Environment namespace | `ServiceAccount`     | Uses namespace of environment                                                                              | Deploying to a cluster     |
+| Environment namespace | `Secret`             | Token for environment ServiceAccount                                                                       | Deploying to a cluster     |
+| Environment namespace | `RoleBinding`        | [`edit`](https://kubernetes.io/docs/reference/access-authn-authz/rbac/#user-facing-roles) roleRef          | Deploying to a cluster     |
+
+NOTE: **Note:**
+Environment-specific resources are only created if your cluster is [managed by GitLab](#gitlab-managed-clusters).
 
 NOTE: **Note:**
-Project-specific resources are only created if your cluster is [managed by GitLab](#gitlab-managed-clusters).
+If your project was created before GitLab 12.2 it will use a single namespace for all project environments.
 
 #### Security of GitLab Runners
 
@@ -640,8 +641,8 @@ GitLab CI/CD build environment.
 | Variable | Description |
 | -------- | ----------- |
 | `KUBE_URL` | Equal to the API URL. |
-| `KUBE_TOKEN` | The Kubernetes token of the [project service account](#access-controls). |
-| `KUBE_NAMESPACE` | The Kubernetes namespace is auto-generated if not specified. The default value is `<project_name>-<project_id>`. You can overwrite it to use different one if needed, otherwise the `KUBE_NAMESPACE` variable will receive the default value. |
+| `KUBE_TOKEN` | The Kubernetes token of the [environment service account](#access-controls). |
+| `KUBE_NAMESPACE` | The Kubernetes namespace is auto-generated if not specified. The default value is `<project_name>-<project_id>-<environment>`. You can overwrite it to use different one if needed, otherwise the `KUBE_NAMESPACE` variable will receive the default value. |
 | `KUBE_CA_PEM_FILE` | Path to a file containing PEM data. Only present if a custom CA bundle was specified. |
 | `KUBE_CA_PEM` | (**deprecated**) Raw PEM data. Only if a custom CA bundle was specified. |
 | `KUBECONFIG` | Path to a file containing `kubeconfig` for this deployment. CA bundle would be embedded if specified. This config also embeds the same token defined in `KUBE_TOKEN` so you likely will only need this variable. This variable name is also automatically picked up by `kubectl` so you won't actually need to reference it explicitly if using `kubectl`. |
diff --git a/doc/user/project/clusters/serverless/index.md b/doc/user/project/clusters/serverless/index.md
index 92ad49e9448..bcf9a677a40 100644
--- a/doc/user/project/clusters/serverless/index.md
+++ b/doc/user/project/clusters/serverless/index.md
@@ -434,7 +434,7 @@ The instructions below relate to installing and running Certbot on a Linux serve
    ./certbot-auto certonly --manual --preferred-challenges dns -d '*.<namespace>.example.com'
    ```
 
-   Where `<namespace>` is the namespace created by GitLab for your serverless project (composed of `<projectname+id>`) and
+   Where `<namespace>` is the namespace created by GitLab for your serverless project (composed of `<project_name>-<project_id>-<environment>`) and
    `example.com` is the domain being used for your project. If you are unsure what the namespace of your project is, navigate
    to the **Operations > Serverless** page of your project and inspect
    the endpoint provided for your function/app.
diff --git a/lib/gitlab/ci/build/prerequisite/kubernetes_namespace.rb b/lib/gitlab/ci/build/prerequisite/kubernetes_namespace.rb
index e6e0aaab60b..6ab4fca3854 100644
--- a/lib/gitlab/ci/build/prerequisite/kubernetes_namespace.rb
+++ b/lib/gitlab/ci/build/prerequisite/kubernetes_namespace.rb
@@ -8,31 +8,51 @@ module Gitlab
           def unmet?
             deployment_cluster.present? &&
               deployment_cluster.managed? &&
-              (kubernetes_namespace.new_record? || kubernetes_namespace.service_account_token.blank?)
+              missing_namespace?
           end
 
           def complete!
             return unless unmet?
 
-            create_or_update_namespace
+            create_namespace
           end
 
           private
 
+          def missing_namespace?
+            kubernetes_namespace.nil? || kubernetes_namespace.service_account_token.blank?
+          end
+
           def deployment_cluster
             build.deployment&.cluster
           end
 
+          def environment
+            build.deployment.environment
+          end
+
           def kubernetes_namespace
             strong_memoize(:kubernetes_namespace) do
-              deployment_cluster.find_or_initialize_kubernetes_namespace_for_project(build.project)
+              Clusters::KubernetesNamespaceFinder.new(
+                deployment_cluster,
+                project: environment.project,
+                environment_slug: environment.slug,
+                allow_blank_token: true
+              ).execute
             end
           end
 
-          def create_or_update_namespace
+          def create_namespace
             Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService.new(
               cluster: deployment_cluster,
-              kubernetes_namespace: kubernetes_namespace
+              kubernetes_namespace: kubernetes_namespace || build_namespace_record
+            ).execute
+          end
+
+          def build_namespace_record
+            Clusters::BuildKubernetesNamespaceService.new(
+              deployment_cluster,
+              environment: environment
             ).execute
           end
         end
diff --git a/lib/gitlab/kubernetes/default_namespace.rb b/lib/gitlab/kubernetes/default_namespace.rb
new file mode 100644
index 00000000000..c95362b024b
--- /dev/null
+++ b/lib/gitlab/kubernetes/default_namespace.rb
@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Kubernetes
+    class DefaultNamespace
+      attr_reader :cluster, :project
+
+      delegate :platform_kubernetes, to: :cluster
+
+      ##
+      # Ideally we would just use an environment record here instead of
+      # passing a project and name/slug separately, but we need to be able
+      # to look up namespaces before the environment has been persisted.
+      def initialize(cluster, project:)
+        @cluster = cluster
+        @project = project
+      end
+
+      def from_environment_name(name)
+        from_environment_slug(generate_slug(name))
+      end
+
+      def from_environment_slug(slug)
+        default_platform_namespace(slug) || default_project_namespace(slug)
+      end
+
+      private
+
+      def default_platform_namespace(slug)
+        return unless platform_kubernetes&.namespace.present?
+
+        if cluster.managed? && cluster.namespace_per_environment?
+          "#{platform_kubernetes.namespace}-#{slug}"
+        else
+          platform_kubernetes.namespace
+        end
+      end
+
+      def default_project_namespace(slug)
+        namespace_slug = "#{project.path}-#{project.id}".downcase
+
+        if cluster.namespace_per_environment?
+          namespace_slug += "-#{slug}"
+        end
+
+        Gitlab::NamespaceSanitizer.sanitize(namespace_slug)
+      end
+
+      ##
+      # Environment slug can be predicted given an environment
+      # name, so even if the environment isn't persisted yet we
+      # still know what to look for.
+      def generate_slug(name)
+        Gitlab::Slug::Environment.new(name).generate
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/prometheus/query_variables.rb b/lib/gitlab/prometheus/query_variables.rb
index 9cc21129547..ba2d33ee1c1 100644
--- a/lib/gitlab/prometheus/query_variables.rb
+++ b/lib/gitlab/prometheus/query_variables.rb
@@ -4,12 +4,9 @@ module Gitlab
   module Prometheus
     module QueryVariables
       def self.call(environment)
-        deployment_platform = environment.deployment_platform
-        namespace = deployment_platform&.kubernetes_namespace_for(environment.project) || ''
-
         {
           ci_environment_slug: environment.slug,
-          kube_namespace: namespace,
+          kube_namespace: environment.deployment_namespace || '',
           environment_filter: %{container_name!="POD",environment="#{environment.slug}"}
         }
       end
diff --git a/spec/controllers/projects/serverless/functions_controller_spec.rb b/spec/controllers/projects/serverless/functions_controller_spec.rb
index 18c594acae0..9f1ef3a4be8 100644
--- a/spec/controllers/projects/serverless/functions_controller_spec.rb
+++ b/spec/controllers/projects/serverless/functions_controller_spec.rb
@@ -10,12 +10,16 @@ describe Projects::Serverless::FunctionsController do
   let(:cluster) { create(:cluster, :project, :provided_by_gcp) }
   let(:service) { cluster.platform_kubernetes }
   let(:project) { cluster.project }
+  let(:environment) { create(:environment, project: project) }
+  let!(:deployment) { create(:deployment, :success, environment: environment, cluster: cluster) }
+  let(:knative_services_finder) { environment.knative_services_finder }
 
   let(:namespace) do
     create(:cluster_kubernetes_namespace,
       cluster: cluster,
       cluster_project: cluster.cluster_project,
-      project: cluster.cluster_project.project)
+      project: cluster.cluster_project.project,
+      environment: environment)
   end
 
   before do
@@ -47,12 +51,11 @@ describe Projects::Serverless::FunctionsController do
     end
 
     context 'when cache is ready' do
-      let(:knative_services_finder) { project.clusters.first.knative_services_finder(project) }
       let(:knative_state) { true }
 
       before do
-        allow_any_instance_of(Clusters::Cluster)
-          .to receive(:knative_services_finder)
+        allow(Clusters::KnativeServicesFinder)
+          .to receive(:new)
           .and_return(knative_services_finder)
         synchronous_reactive_cache(knative_services_finder)
         stub_kubeclient_service_pods(
@@ -107,12 +110,12 @@ describe Projects::Serverless::FunctionsController do
     context 'valid data', :use_clean_rails_memory_store_caching do
       before do
         stub_kubeclient_service_pods
-        stub_reactive_cache(cluster.knative_services_finder(project),
+        stub_reactive_cache(knative_services_finder,
           {
             services: kube_knative_services_body(namespace: namespace.namespace, name: cluster.project.name)["items"],
             pods: kube_knative_pods_body(cluster.project.name, namespace.namespace)["items"]
           },
-          *cluster.knative_services_finder(project).cache_args)
+          *knative_services_finder.cache_args)
       end
 
       it 'has a valid function name' do
@@ -140,12 +143,12 @@ describe Projects::Serverless::FunctionsController do
   describe 'GET #index with data', :use_clean_rails_memory_store_caching do
     before do
       stub_kubeclient_service_pods
-      stub_reactive_cache(cluster.knative_services_finder(project),
+      stub_reactive_cache(knative_services_finder,
         {
           services: kube_knative_services_body(namespace: namespace.namespace, name: cluster.project.name)["items"],
           pods: kube_knative_pods_body(cluster.project.name, namespace.namespace)["items"]
         },
-        *cluster.knative_services_finder(project).cache_args)
+        *knative_services_finder.cache_args)
     end
 
     it 'has data' do
diff --git a/spec/factories/clusters/clusters.rb b/spec/factories/clusters/clusters.rb
index b0d14b672f4..d294e6d055e 100644
--- a/spec/factories/clusters/clusters.rb
+++ b/spec/factories/clusters/clusters.rb
@@ -6,6 +6,7 @@ FactoryBot.define do
     name 'test-cluster'
     cluster_type :project_type
     managed true
+    namespace_per_environment true
 
     factory :cluster_for_group, traits: [:provided_by_gcp, :group]
 
@@ -29,6 +30,10 @@ FactoryBot.define do
       end
     end
 
+    trait :namespace_per_environment_disabled do
+      namespace_per_environment false
+    end
+
     trait :provided_by_user do
       provider_type :user
       platform_type :kubernetes
diff --git a/spec/factories/clusters/kubernetes_namespaces.rb b/spec/factories/clusters/kubernetes_namespaces.rb
index 042be7b4c4a..8d6ad1b9f79 100644
--- a/spec/factories/clusters/kubernetes_namespaces.rb
+++ b/spec/factories/clusters/kubernetes_namespaces.rb
@@ -5,12 +5,21 @@ FactoryBot.define do
     association :cluster, :project, :provided_by_gcp
 
     after(:build) do |kubernetes_namespace|
-      if kubernetes_namespace.cluster.project_type?
-        cluster_project = kubernetes_namespace.cluster.cluster_project
+      cluster = kubernetes_namespace.cluster
+
+      if cluster.project_type?
+        cluster_project = cluster.cluster_project
 
         kubernetes_namespace.project = cluster_project.project
         kubernetes_namespace.cluster_project = cluster_project
       end
+
+      kubernetes_namespace.namespace ||=
+        Gitlab::Kubernetes::DefaultNamespace.new(
+          cluster,
+          project: kubernetes_namespace.project
+        ).from_environment_slug(kubernetes_namespace.environment&.slug)
+      kubernetes_namespace.service_account_name ||= "#{kubernetes_namespace.namespace}-service-account"
     end
 
     trait :with_token do
diff --git a/spec/features/projects/serverless/functions_spec.rb b/spec/features/projects/serverless/functions_spec.rb
index 9865dbbfb3c..e82e5b81021 100644
--- a/spec/features/projects/serverless/functions_spec.rb
+++ b/spec/features/projects/serverless/functions_spec.rb
@@ -39,17 +39,19 @@ describe 'Functions', :js do
     let(:cluster) { create(:cluster, :project, :provided_by_gcp) }
     let(:service) { cluster.platform_kubernetes }
     let(:project) { cluster.project }
-    let(:knative_services_finder) { project.clusters.first.knative_services_finder(project) }
+    let(:environment) { create(:environment, project: project) }
+    let!(:deployment) { create(:deployment, :success, cluster: cluster, environment: environment) }
+    let(:knative_services_finder) { environment.knative_services_finder }
     let(:namespace) do
       create(:cluster_kubernetes_namespace,
         cluster: cluster,
-        cluster_project: cluster.cluster_project,
-        project: cluster.cluster_project.project)
+        project: cluster.cluster_project.project,
+        environment: environment)
     end
 
     before do
-      allow_any_instance_of(Clusters::Cluster)
-        .to receive(:knative_services_finder)
+      allow(Clusters::KnativeServicesFinder)
+        .to receive(:new)
         .and_return(knative_services_finder)
       synchronous_reactive_cache(knative_services_finder)
       stub_kubeclient_knative_services(stub_get_services_options)
diff --git a/spec/finders/clusters/knative_services_finder_spec.rb b/spec/finders/clusters/knative_services_finder_spec.rb
index b731c2bd6bf..159724b3c1f 100644
--- a/spec/finders/clusters/knative_services_finder_spec.rb
+++ b/spec/finders/clusters/knative_services_finder_spec.rb
@@ -7,15 +7,19 @@ describe Clusters::KnativeServicesFinder do
   include ReactiveCachingHelpers
 
   let(:cluster) { create(:cluster, :project, :provided_by_gcp) }
-  let(:service) { cluster.platform_kubernetes }
+  let(:service) { environment.deployment_platform }
   let(:project) { cluster.cluster_project.project }
+  let(:environment) { create(:environment, project: project) }
+  let!(:deployment) { create(:deployment, :success, environment: environment, cluster: cluster) }
   let(:namespace) do
     create(:cluster_kubernetes_namespace,
       cluster: cluster,
-      cluster_project: cluster.cluster_project,
-      project: project)
+      project: project,
+      environment: environment)
   end
 
+  let(:finder) { described_class.new(cluster, environment) }
+
   before do
     stub_kubeclient_knative_services(namespace: namespace.namespace)
     stub_kubeclient_service_pods(
@@ -35,7 +39,7 @@ describe Clusters::KnativeServicesFinder do
 
     context 'when using synchronous reactive cache' do
       before do
-        synchronous_reactive_cache(cluster.knative_services_finder(project))
+        synchronous_reactive_cache(finder)
       end
 
       context 'when there are functions for cluster namespace' do
@@ -60,21 +64,21 @@ describe Clusters::KnativeServicesFinder do
   end
 
   describe '#service_pod_details' do
-    subject { cluster.knative_services_finder(project).service_pod_details(project.name) }
+    subject { finder.service_pod_details(project.name) }
 
     it_behaves_like 'a cached data'
   end
 
   describe '#services' do
-    subject { cluster.knative_services_finder(project).services }
+    subject { finder.services }
 
     it_behaves_like 'a cached data'
   end
 
   describe '#knative_detected' do
-    subject { cluster.knative_services_finder(project).knative_detected }
+    subject { finder.knative_detected }
     before do
-      synchronous_reactive_cache(cluster.knative_services_finder(project))
+      synchronous_reactive_cache(finder)
     end
 
     context 'when knative is installed' do
@@ -85,7 +89,7 @@ describe Clusters::KnativeServicesFinder do
       it { is_expected.to be_truthy }
       it "discovers knative installation" do
         expect { subject }
-          .to change { cluster.kubeclient.knative_client.discovered }
+          .to change { finder.cluster.kubeclient.knative_client.discovered }
           .from(false)
           .to(true)
       end
diff --git a/spec/finders/clusters/kubernetes_namespace_finder_spec.rb b/spec/finders/clusters/kubernetes_namespace_finder_spec.rb
new file mode 100644
index 00000000000..8beba0b99a4
--- /dev/null
+++ b/spec/finders/clusters/kubernetes_namespace_finder_spec.rb
@@ -0,0 +1,110 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Clusters::KubernetesNamespaceFinder do
+  let(:finder) do
+    described_class.new(
+      cluster,
+      project: project,
+      environment_slug: 'production',
+      allow_blank_token: allow_blank_token
+    )
+  end
+
+  def create_namespace(environment, with_token: true)
+    create(:cluster_kubernetes_namespace,
+      (with_token ? :with_token : :without_token),
+      cluster: cluster,
+      project: project,
+      environment: environment
+    )
+  end
+
+  describe '#execute' do
+    let(:production) { create(:environment, project: project, slug: 'production') }
+    let(:staging) { create(:environment, project: project, slug: 'staging') }
+
+    let(:cluster) { create(:cluster, :group, :provided_by_user) }
+    let(:project) { create(:project) }
+    let(:allow_blank_token) { false }
+
+    subject { finder.execute }
+
+    before do
+      allow(cluster).to receive(:namespace_per_environment?).and_return(namespace_per_environment)
+    end
+
+    context 'cluster supports separate namespaces per environment' do
+      let(:namespace_per_environment) { true }
+
+      context 'no persisted namespace is present' do
+        it { is_expected.to be_nil }
+      end
+
+      context 'a namespace with an environment is present' do
+        context 'environment matches' do
+          let!(:namespace_with_environment) { create_namespace(production) }
+
+          it { is_expected.to eq namespace_with_environment }
+
+          context 'project cluster' do
+            let(:cluster) { create(:cluster, :project, :provided_by_user, projects: [project]) }
+
+            it { is_expected.to eq namespace_with_environment }
+          end
+
+          context 'service account token is blank' do
+            let!(:namespace_with_environment) { create_namespace(production, with_token: false) }
+
+            it { is_expected.to be_nil }
+
+            context 'allow_blank_token is true' do
+              let(:allow_blank_token) { true }
+
+              it { is_expected.to eq namespace_with_environment }
+            end
+          end
+        end
+
+        context 'environment does not match' do
+          let!(:namespace_with_environment) { create_namespace(staging) }
+
+          it { is_expected.to be_nil }
+        end
+      end
+    end
+
+    context 'cluster does not support separate namespaces per environment' do
+      let(:namespace_per_environment) { false }
+
+      context 'no persisted namespace is present' do
+        it { is_expected.to be_nil }
+      end
+
+      context 'a legacy namespace with no environment is present' do
+        let!(:legacy_namespace) { create_namespace(nil) }
+
+        it { is_expected.to eq legacy_namespace }
+
+        context 'project cluster' do
+          let(:cluster) { create(:cluster, :project, :provided_by_user, projects: [project]) }
+
+          it { is_expected.to eq legacy_namespace }
+        end
+
+        context 'service account token is blank' do
+          let!(:legacy_namespace) { create_namespace(nil, with_token: false) }
+
+          it { is_expected.to be_nil }
+
+          context 'allow_blank_token is true' do
+            let(:allow_blank_token) { true }
+
+            it { is_expected.to eq legacy_namespace }
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/finders/projects/serverless/functions_finder_spec.rb b/spec/finders/projects/serverless/functions_finder_spec.rb
index 8aea45b457c..589e4000d46 100644
--- a/spec/finders/projects/serverless/functions_finder_spec.rb
+++ b/spec/finders/projects/serverless/functions_finder_spec.rb
@@ -11,12 +11,15 @@ describe Projects::Serverless::FunctionsFinder do
   let(:cluster) { create(:cluster, :project, :provided_by_gcp) }
   let(:service) { cluster.platform_kubernetes }
   let(:project) { cluster.project }
+  let(:environment) { create(:environment, project: project) }
+  let!(:deployment) { create(:deployment, :success, environment: environment, cluster: cluster) }
+  let(:knative_services_finder) { environment.knative_services_finder }
 
   let(:namespace) do
     create(:cluster_kubernetes_namespace,
       cluster: cluster,
-      cluster_project: cluster.cluster_project,
-      project: cluster.cluster_project.project)
+      project: project,
+      environment: environment)
   end
 
   before do
@@ -29,11 +32,9 @@ describe Projects::Serverless::FunctionsFinder do
     end
 
     context 'when reactive_caching has finished' do
-      let(:knative_services_finder) { project.clusters.first.knative_services_finder(project) }
-
       before do
-        allow_any_instance_of(Clusters::Cluster)
-          .to receive(:knative_services_finder)
+        allow(Clusters::KnativeServicesFinder)
+          .to receive(:new)
           .and_return(knative_services_finder)
         synchronous_reactive_cache(knative_services_finder)
       end
@@ -47,8 +48,6 @@ describe Projects::Serverless::FunctionsFinder do
       end
 
       context 'reactive_caching is finished and knative is installed' do
-        let(:knative_services_finder) { project.clusters.first.knative_services_finder(project) }
-
         it 'returns true' do
           stub_kubeclient_knative_services(namespace: namespace.namespace)
           stub_kubeclient_service_pods(nil, namespace: namespace.namespace)
@@ -74,24 +73,24 @@ describe Projects::Serverless::FunctionsFinder do
 
       it 'there are functions', :use_clean_rails_memory_store_caching do
         stub_kubeclient_service_pods
-        stub_reactive_cache(cluster.knative_services_finder(project),
+        stub_reactive_cache(knative_services_finder,
           {
             services: kube_knative_services_body(namespace: namespace.namespace, name: cluster.project.name)["items"],
             pods: kube_knative_pods_body(cluster.project.name, namespace.namespace)["items"]
           },
-          *cluster.knative_services_finder(project).cache_args)
+          *knative_services_finder.cache_args)
 
         expect(finder.execute).not_to be_empty
       end
 
       it 'has a function', :use_clean_rails_memory_store_caching do
         stub_kubeclient_service_pods
-        stub_reactive_cache(cluster.knative_services_finder(project),
+        stub_reactive_cache(knative_services_finder,
           {
             services: kube_knative_services_body(namespace: namespace.namespace, name: cluster.project.name)["items"],
             pods: kube_knative_pods_body(cluster.project.name, namespace.namespace)["items"]
           },
-          *cluster.knative_services_finder(project).cache_args)
+          *knative_services_finder.cache_args)
 
         result = finder.service(cluster.environment_scope, cluster.project.name)
         expect(result).not_to be_empty
@@ -109,7 +108,7 @@ describe Projects::Serverless::FunctionsFinder do
       let(:finder) { described_class.new(project) }
 
       before do
-        allow(finder).to receive(:prometheus_adapter).and_return(prometheus_adapter)
+        allow(Prometheus::AdapterService).to receive(:new).and_return(double(prometheus_adapter: prometheus_adapter))
         allow(prometheus_adapter).to receive(:query).and_return(prometheus_empty_body('matrix'))
       end
 
diff --git a/spec/lib/gitlab/ci/build/prerequisite/kubernetes_namespace_spec.rb b/spec/lib/gitlab/ci/build/prerequisite/kubernetes_namespace_spec.rb
index d88a2097ba2..775550f2acc 100644
--- a/spec/lib/gitlab/ci/build/prerequisite/kubernetes_namespace_spec.rb
+++ b/spec/lib/gitlab/ci/build/prerequisite/kubernetes_namespace_spec.rb
@@ -3,9 +3,9 @@
 require 'spec_helper'
 
 describe Gitlab::Ci::Build::Prerequisite::KubernetesNamespace do
-  let(:build) { create(:ci_build) }
-
   describe '#unmet?' do
+    let(:build) { create(:ci_build) }
+
     subject { described_class.new(build).unmet? }
 
     context 'build has no deployment' do
@@ -18,7 +18,6 @@ describe Gitlab::Ci::Build::Prerequisite::KubernetesNamespace do
 
     context 'build has a deployment' do
       let!(:deployment) { create(:deployment, deployable: build, cluster: cluster) }
-      let(:cluster) { nil }
 
       context 'and a cluster to deploy to' do
         let(:cluster) { create(:cluster, :group) }
@@ -32,12 +31,17 @@ describe Gitlab::Ci::Build::Prerequisite::KubernetesNamespace do
         end
 
         context 'and a namespace is already created for this project' do
-          let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token, cluster: cluster, project: build.project) }
+          let(:kubernetes_namespace) { instance_double(Clusters::KubernetesNamespace, service_account_token: 'token') }
+
+          before do
+            allow(Clusters::KubernetesNamespaceFinder).to receive(:new)
+              .and_return(double(execute: kubernetes_namespace))
+          end
 
           it { is_expected.to be_falsey }
 
           context 'and the service_account_token is blank' do
-            let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :without_token, cluster: cluster, project: build.project) }
+            let(:kubernetes_namespace) { instance_double(Clusters::KubernetesNamespace, service_account_token: nil) }
 
             it { is_expected.to be_truthy }
           end
@@ -45,34 +49,79 @@ describe Gitlab::Ci::Build::Prerequisite::KubernetesNamespace do
       end
 
       context 'and no cluster to deploy to' do
+        let(:cluster) { nil }
+
         it { is_expected.to be_falsey }
       end
     end
   end
 
   describe '#complete!' do
-    let!(:deployment) { create(:deployment, deployable: build, cluster: cluster) }
-    let(:service) { double(execute: true) }
-    let(:cluster) { nil }
+    let(:build) { create(:ci_build) }
+    let(:prerequisite) { described_class.new(build) }
 
-    subject { described_class.new(build).complete! }
+    subject { prerequisite.complete! }
 
     context 'completion is required' do
       let(:cluster) { create(:cluster, :group) }
+      let(:deployment) { create(:deployment, cluster: cluster) }
+      let(:service) { double(execute: true) }
+      let(:kubernetes_namespace) { double }
+
+      before do
+        allow(prerequisite).to receive(:unmet?).and_return(true)
+        allow(build).to receive(:deployment).and_return(deployment)
+      end
+
+      context 'kubernetes namespace does not exist' do
+        let(:namespace_builder) { double(execute: kubernetes_namespace)}
 
-      it 'creates a kubernetes namespace' do
-        expect(Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService)
-          .to receive(:new)
-          .with(cluster: cluster, kubernetes_namespace: instance_of(Clusters::KubernetesNamespace))
-          .and_return(service)
+        before do
+          allow(Clusters::KubernetesNamespaceFinder).to receive(:new)
+            .and_return(double(execute: nil))
+        end
 
-        expect(service).to receive(:execute).once
+        it 'creates a namespace using a new record' do
+          expect(Clusters::BuildKubernetesNamespaceService)
+            .to receive(:new)
+            .with(cluster, environment: deployment.environment)
+            .and_return(namespace_builder)
 
-        subject
+          expect(Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService)
+            .to receive(:new)
+            .with(cluster: cluster, kubernetes_namespace: kubernetes_namespace)
+            .and_return(service)
+
+          expect(service).to receive(:execute).once
+
+          subject
+        end
+      end
+
+      context 'kubernetes namespace exists (but has no service_account_token)' do
+        before do
+          allow(Clusters::KubernetesNamespaceFinder).to receive(:new)
+            .and_return(double(execute: kubernetes_namespace))
+        end
+
+        it 'creates a namespace using the tokenless record' do
+          expect(Clusters::BuildKubernetesNamespaceService).not_to receive(:new)
+
+          expect(Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService)
+            .to receive(:new)
+            .with(cluster: cluster, kubernetes_namespace: kubernetes_namespace)
+            .and_return(service)
+
+          subject
+        end
       end
     end
 
     context 'completion is not required' do
+      before do
+        allow(prerequisite).to receive(:unmet?).and_return(false)
+      end
+
       it 'does not create a namespace' do
         expect(Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService).not_to receive(:new)
 
diff --git a/spec/lib/gitlab/kubernetes/default_namespace_spec.rb b/spec/lib/gitlab/kubernetes/default_namespace_spec.rb
new file mode 100644
index 00000000000..1fda547f35c
--- /dev/null
+++ b/spec/lib/gitlab/kubernetes/default_namespace_spec.rb
@@ -0,0 +1,85 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Gitlab::Kubernetes::DefaultNamespace do
+  let(:generator) { described_class.new(cluster, project: environment.project) }
+
+  describe '#from_environment_name' do
+    let(:cluster) { create(:cluster) }
+    let(:environment) { create(:environment) }
+
+    subject { generator.from_environment_name(environment.name) }
+
+    it 'generates a slug and passes it to #from_environment_slug' do
+      expect(Gitlab::Slug::Environment).to receive(:new)
+        .with(environment.name)
+        .and_return(double(generate: environment.slug))
+
+      expect(generator).to receive(:from_environment_slug)
+        .with(environment.slug)
+        .and_return(:mock_namespace)
+
+      expect(subject).to eq :mock_namespace
+    end
+  end
+
+  describe '#from_environment_slug' do
+    let(:platform) { create(:cluster_platform_kubernetes, namespace: platform_namespace) }
+    let(:cluster) { create(:cluster, platform_kubernetes: platform) }
+    let(:project) { create(:project, path: "Path-With-Capitals") }
+    let(:environment) { create(:environment, project: project) }
+
+    subject { generator.from_environment_slug(environment.slug) }
+
+    context 'namespace per environment is enabled' do
+      context 'platform namespace is specified' do
+        let(:platform_namespace) { 'platform-namespace' }
+
+        it { is_expected.to eq "#{platform_namespace}-#{environment.slug}" }
+
+        context 'cluster is unmanaged' do
+          let(:cluster) { create(:cluster, :not_managed, platform_kubernetes: platform) }
+
+          it { is_expected.to eq platform_namespace }
+        end
+      end
+
+      context 'platform namespace is blank' do
+        let(:platform_namespace) { nil }
+        let(:mock_namespace) { 'mock-namespace' }
+
+        it 'constructs a namespace from the project and environment' do
+          expect(Gitlab::NamespaceSanitizer).to receive(:sanitize)
+            .with("#{project.path}-#{project.id}-#{environment.slug}".downcase)
+            .and_return(mock_namespace)
+
+          expect(subject).to eq mock_namespace
+        end
+      end
+    end
+
+    context 'namespace per environment is disabled' do
+      let(:cluster) { create(:cluster, :namespace_per_environment_disabled, platform_kubernetes: platform) }
+
+      context 'platform namespace is specified' do
+        let(:platform_namespace) { 'platform-namespace' }
+
+        it { is_expected.to eq platform_namespace }
+      end
+
+      context 'platform namespace is blank' do
+        let(:platform_namespace) { nil }
+        let(:mock_namespace) { 'mock-namespace' }
+
+        it 'constructs a namespace from the project and environment' do
+          expect(Gitlab::NamespaceSanitizer).to receive(:sanitize)
+            .with("#{project.path}-#{project.id}".downcase)
+            .and_return(mock_namespace)
+
+          expect(subject).to eq mock_namespace
+        end
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/prometheus/query_variables_spec.rb b/spec/lib/gitlab/prometheus/query_variables_spec.rb
index 6dc99ef26ec..3f9b245a3fb 100644
--- a/spec/lib/gitlab/prometheus/query_variables_spec.rb
+++ b/spec/lib/gitlab/prometheus/query_variables_spec.rb
@@ -23,7 +23,7 @@ describe Gitlab::Prometheus::QueryVariables do
 
     context 'with deployment platform' do
       context 'with project cluster' do
-        let(:kube_namespace) { environment.deployment_platform.cluster.kubernetes_namespace_for(project) }
+        let(:kube_namespace) { environment.deployment_namespace }
 
         before do
           create(:cluster, :project, :provided_by_user, projects: [project])
@@ -38,8 +38,8 @@ describe Gitlab::Prometheus::QueryVariables do
         let(:project2) { create(:project) }
         let(:kube_namespace) { k8s_ns.namespace }
 
-        let!(:k8s_ns) { create(:cluster_kubernetes_namespace, cluster: cluster, project: project) }
-        let!(:k8s_ns2) { create(:cluster_kubernetes_namespace, cluster: cluster, project: project2) }
+        let!(:k8s_ns) { create(:cluster_kubernetes_namespace, cluster: cluster, project: project, environment: environment) }
+        let!(:k8s_ns2) { create(:cluster_kubernetes_namespace, cluster: cluster, project: project2, environment: environment) }
 
         before do
           group.projects << project
diff --git a/spec/models/clusters/cluster_spec.rb b/spec/models/clusters/cluster_spec.rb
index 96212d0c864..9afbe6328ca 100644
--- a/spec/models/clusters/cluster_spec.rb
+++ b/spec/models/clusters/cluster_spec.rb
@@ -38,11 +38,6 @@ describe Clusters::Cluster, :use_clean_rails_memory_store_caching do
 
   it { is_expected.to respond_to :project }
 
-  it do
-    expect(subject.knative_services_finder(subject.project))
-      .to be_instance_of(Clusters::KnativeServicesFinder)
-  end
-
   describe '.enabled' do
     subject { described_class.enabled }
 
@@ -534,60 +529,39 @@ describe Clusters::Cluster, :use_clean_rails_memory_store_caching do
     end
   end
 
-  describe '#find_or_initialize_kubernetes_namespace_for_project' do
-    let(:cluster) { create(:cluster, :project, :provided_by_gcp) }
-    let(:project) { cluster.projects.first }
-
-    subject { cluster.find_or_initialize_kubernetes_namespace_for_project(project) }
-
-    context 'kubernetes namespace exists' do
-      context 'with no service account token' do
-        let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, project: project, cluster: cluster) }
-
-        it { is_expected.to eq kubernetes_namespace }
-      end
+  describe '#kubernetes_namespace_for' do
+    let(:cluster) { create(:cluster, :group) }
+    let(:environment) { create(:environment) }
 
-      context 'with a service account token' do
-        let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token, project: project, cluster: cluster) }
+    subject { cluster.kubernetes_namespace_for(environment) }
 
-        it { is_expected.to eq kubernetes_namespace }
-      end
-    end
-
-    context 'kubernetes namespace does not exist' do
-      it 'initializes a new namespace and sets default values' do
-        expect(subject).to be_new_record
-        expect(subject.project).to eq project
-        expect(subject.cluster).to eq cluster
-        expect(subject.namespace).to be_present
-        expect(subject.service_account_name).to be_present
-      end
+    before do
+      expect(Clusters::KubernetesNamespaceFinder).to receive(:new)
+        .with(cluster, project: environment.project, environment_slug: environment.slug)
+        .and_return(double(execute: persisted_namespace))
     end
 
-    context 'a custom scope is provided' do
-      let(:scope) { cluster.kubernetes_namespaces.has_service_account_token }
-
-      subject { cluster.find_or_initialize_kubernetes_namespace_for_project(project, scope: scope) }
-
-      context 'kubernetes namespace exists' do
-        context 'with no service account token' do
-          let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, project: project, cluster: cluster) }
+    context 'a persisted namespace exists' do
+      let(:persisted_namespace) { create(:cluster_kubernetes_namespace) }
 
-          it 'initializes a new namespace and sets default values' do
-            expect(subject).to be_new_record
-            expect(subject.project).to eq project
-            expect(subject.cluster).to eq cluster
-            expect(subject.namespace).to be_present
-            expect(subject.service_account_name).to be_present
-          end
-        end
+      it { is_expected.to eq persisted_namespace.namespace }
+    end
 
-        context 'with a service account token' do
-          let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token, project: project, cluster: cluster) }
+    context 'no persisted namespace exists' do
+      let(:persisted_namespace) { nil }
+      let(:namespace_generator) { double }
+      let(:default_namespace) { 'a-default-namespace' }
 
-          it { is_expected.to eq kubernetes_namespace }
-        end
+      before do
+        expect(Gitlab::Kubernetes::DefaultNamespace).to receive(:new)
+          .with(cluster, project: environment.project)
+          .and_return(namespace_generator)
+        expect(namespace_generator).to receive(:from_environment_slug)
+          .with(environment.slug)
+          .and_return(default_namespace)
       end
+
+      it { is_expected.to eq default_namespace }
     end
   end
 
diff --git a/spec/models/clusters/kubernetes_namespace_spec.rb b/spec/models/clusters/kubernetes_namespace_spec.rb
index b5cba80b806..d4e3a0ac84d 100644
--- a/spec/models/clusters/kubernetes_namespace_spec.rb
+++ b/spec/models/clusters/kubernetes_namespace_spec.rb
@@ -24,70 +24,60 @@ RSpec.describe Clusters::KubernetesNamespace, type: :model do
     end
   end
 
-  describe 'namespace uniqueness validation' do
-    let(:cluster_project) { create(:cluster_project) }
-    let(:kubernetes_namespace) { build(:cluster_kubernetes_namespace, namespace: 'my-namespace') }
+  describe '.with_environment_slug' do
+    let(:cluster) { create(:cluster, :group) }
+    let(:environment) { create(:environment, slug: slug) }
 
-    subject { kubernetes_namespace }
+    let(:slug) { 'production' }
 
-    context 'when cluster is using the namespace' do
-      before do
-        create(:cluster_kubernetes_namespace,
-               cluster: kubernetes_namespace.cluster,
-               namespace: 'my-namespace')
-      end
+    subject { described_class.with_environment_slug(slug) }
 
-      it { is_expected.not_to be_valid }
-    end
+    context 'there is no associated environment' do
+      let!(:namespace) { create(:cluster_kubernetes_namespace, cluster: cluster, project: environment.project) }
 
-    context 'when cluster is not using the namespace' do
-      it { is_expected.to be_valid }
+      it { is_expected.to be_empty }
     end
-  end
 
-  describe '#set_defaults' do
-    let(:kubernetes_namespace) { build(:cluster_kubernetes_namespace) }
-    let(:cluster) { kubernetes_namespace.cluster }
-    let(:platform) { kubernetes_namespace.platform_kubernetes }
-
-    subject { kubernetes_namespace.set_defaults }
-
-    describe '#namespace' do
-      before do
-        platform.update_column(:namespace, namespace)
+    context 'there is an assicated environment' do
+      let!(:namespace) do
+        create(
+          :cluster_kubernetes_namespace,
+          cluster: cluster,
+          project: environment.project,
+          environment: environment
+        )
       end
 
-      context 'when platform has a namespace assigned' do
-        let(:namespace) { 'platform-namespace' }
-
-        it 'copies the namespace' do
-          subject
-
-          expect(kubernetes_namespace.namespace).to eq('platform-namespace')
-        end
+      context 'with a matching slug' do
+        it { is_expected.to eq [namespace] }
       end
 
-      context 'when platform does not have namespace assigned' do
-        let(:project) { kubernetes_namespace.project }
-        let(:namespace) { nil }
-        let(:project_slug) { "#{project.path}-#{project.id}" }
-
-        it 'fallbacks to project namespace' do
-          subject
+      context 'without a matching slug' do
+        let(:environment) { create(:environment, slug: 'staging') }
 
-          expect(kubernetes_namespace.namespace).to eq(project_slug)
-        end
+        it { is_expected.to be_empty }
       end
     end
+  end
 
-    describe '#service_account_name' do
-      let(:service_account_name) { "#{kubernetes_namespace.namespace}-service-account" }
+  describe 'namespace uniqueness validation' do
+    let(:kubernetes_namespace) { build(:cluster_kubernetes_namespace, namespace: 'my-namespace') }
 
-      it 'sets a service account name based on namespace' do
-        subject
+    subject { kubernetes_namespace }
 
-        expect(kubernetes_namespace.service_account_name).to eq(service_account_name)
+    context 'when cluster is using the namespace' do
+      before do
+        create(:cluster_kubernetes_namespace,
+               cluster: kubernetes_namespace.cluster,
+               environment: kubernetes_namespace.environment,
+               namespace: 'my-namespace')
       end
+
+      it { is_expected.not_to be_valid }
+    end
+
+    context 'when cluster is not using the namespace' do
+      it { is_expected.to be_valid }
     end
   end
 
diff --git a/spec/models/clusters/platforms/kubernetes_spec.rb b/spec/models/clusters/platforms/kubernetes_spec.rb
index 5811016ea4d..0c4cf291d20 100644
--- a/spec/models/clusters/platforms/kubernetes_spec.rb
+++ b/spec/models/clusters/platforms/kubernetes_spec.rb
@@ -205,192 +205,77 @@ describe Clusters::Platforms::Kubernetes do
     it { is_expected.to be_truthy }
   end
 
-  describe '#kubernetes_namespace_for' do
-    let(:cluster) { create(:cluster, :project) }
-    let(:project) { cluster.project }
-
-    let(:platform) do
-      create(:cluster_platform_kubernetes,
-             cluster: cluster,
-             namespace: namespace)
-    end
-
-    subject { platform.kubernetes_namespace_for(project) }
-
-    context 'with a namespace assigned' do
-      let(:namespace) { 'namespace-123' }
-
-      it { is_expected.to eq(namespace) }
-
-      context 'kubernetes namespace is present but has no service account token' do
-        let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, cluster: cluster) }
-
-        it { is_expected.to eq(namespace) }
-      end
-    end
-
-    context 'with no namespace assigned' do
-      let(:namespace) { nil }
-
-      context 'when kubernetes namespace is present' do
-        let(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token, cluster: cluster) }
-
-        before do
-          kubernetes_namespace
-        end
-
-        it { is_expected.to eq(kubernetes_namespace.namespace) }
-
-        context 'kubernetes namespace has no service account token' do
-          before do
-            kubernetes_namespace.update!(namespace: 'old-namespace', service_account_token: nil)
-          end
+  describe '#predefined_variables' do
+    let(:project) { create(:project) }
+    let(:cluster) { create(:cluster, :group, platform_kubernetes: platform) }
+    let(:platform) { create(:cluster_platform_kubernetes) }
+    let(:persisted_namespace) { create(:cluster_kubernetes_namespace, project: project, cluster: cluster) }
 
-          it { is_expected.to eq("#{project.path}-#{project.id}") }
-        end
-      end
+    let(:environment_name) { 'env/production' }
+    let(:environment_slug) { Gitlab::Slug::Environment.new(environment_name).generate }
 
-      context 'when kubernetes namespace is not present' do
-        it { is_expected.to eq("#{project.path}-#{project.id}") }
-      end
-    end
-  end
+    subject { platform.predefined_variables(project: project, environment_name: environment_name) }
 
-  describe '#predefined_variables' do
-    let!(:cluster) { create(:cluster, :project, platform_kubernetes: kubernetes) }
-    let(:kubernetes) { create(:cluster_platform_kubernetes, api_url: api_url, ca_cert: ca_pem) }
-    let(:api_url) { 'https://kube.domain.com' }
-    let(:ca_pem) { File.read(Rails.root.join('spec/fixtures/clusters/sample_cert.pem')) }
-
-    subject { kubernetes.predefined_variables(project: cluster.project) }
-
-    shared_examples 'setting variables' do
-      it 'sets the variables' do
-        expect(subject).to include(
-          { key: 'KUBE_URL', value: api_url, public: true },
-          { key: 'KUBE_CA_PEM', value: ca_pem, public: true },
-          { key: 'KUBE_CA_PEM_FILE', value: ca_pem, public: true, file: true }
-        )
-      end
+    before do
+      allow(Clusters::KubernetesNamespaceFinder).to receive(:new)
+        .with(cluster, project: project, environment_slug: environment_slug)
+        .and_return(double(execute: persisted_namespace))
     end
 
-    context 'kubernetes namespace is created with no service account token' do
-      let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, cluster: cluster) }
+    it { is_expected.to include(key: 'KUBE_URL', value: platform.api_url, public: true) }
 
-      it_behaves_like 'setting variables'
+    context 'platform has a CA certificate' do
+      let(:ca_pem) { File.read(Rails.root.join('spec/fixtures/clusters/sample_cert.pem')) }
+      let(:platform) { create(:cluster_platform_kubernetes, ca_cert: ca_pem) }
 
-      it 'does not set KUBE_TOKEN' do
-        expect(subject).not_to include(
-          { key: 'KUBE_TOKEN', value: kubernetes.token, public: false, masked: true }
-        )
-      end
+      it { is_expected.to include(key: 'KUBE_CA_PEM', value: ca_pem, public: true) }
+      it { is_expected.to include(key: 'KUBE_CA_PEM_FILE', value: ca_pem, public: true, file: true) }
     end
 
-    context 'kubernetes namespace is created with service account token' do
-      let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token, cluster: cluster) }
-
-      it_behaves_like 'setting variables'
+    context 'kubernetes namespace exists' do
+      let(:variable) { Hash(key: :fake_key, value: 'fake_value') }
+      let(:namespace_variables) { Gitlab::Ci::Variables::Collection.new([variable]) }
 
-      it 'sets KUBE_TOKEN' do
-        expect(subject).to include(
-          { key: 'KUBE_TOKEN', value: kubernetes_namespace.service_account_token, public: false, masked: true }
-        )
+      before do
+        expect(persisted_namespace).to receive(:predefined_variables).and_return(namespace_variables)
       end
 
-      context 'the cluster has been set to unmanaged after the namespace was created' do
-        before do
-          cluster.update!(managed: false)
-        end
-
-        it_behaves_like 'setting variables'
-
-        it 'sets KUBE_TOKEN from the platform' do
-          expect(subject).to include(
-            { key: 'KUBE_TOKEN', value: kubernetes.token, public: false, masked: true }
-          )
-        end
-
-        context 'the platform has a custom namespace set' do
-          before do
-            kubernetes.update!(namespace: 'custom-namespace')
-          end
-
-          it 'sets KUBE_NAMESPACE from the platform' do
-            expect(subject).to include(
-              { key: 'KUBE_NAMESPACE', value: kubernetes.namespace, public: true, masked: false }
-            )
-          end
-        end
-
-        context 'there is no namespace specified on the platform' do
-          let(:project) { cluster.project }
-
-          before do
-            kubernetes.update!(namespace: nil)
-          end
-
-          it 'sets KUBE_NAMESPACE to a default for the project' do
-            expect(subject).to include(
-              { key: 'KUBE_NAMESPACE', value: "#{project.path}-#{project.id}", public: true, masked: false }
-            )
-          end
-        end
-      end
+      it { is_expected.to include(variable) }
     end
 
-    context 'group level cluster' do
-      let!(:cluster) { create(:cluster, :group, platform_kubernetes: kubernetes) }
-
-      let(:project) { create(:project, group: cluster.group) }
-
-      subject { kubernetes.predefined_variables(project: project) }
-
-      context 'no kubernetes namespace for the project' do
-        it_behaves_like 'setting variables'
-
-        it 'does not return KUBE_TOKEN' do
-          expect(subject).not_to include(
-            { key: 'KUBE_TOKEN', value: kubernetes.token, public: false }
-          )
-        end
-
-        context 'the cluster is not managed' do
-          let!(:cluster) { create(:cluster, :group, :not_managed, platform_kubernetes: kubernetes) }
+    context 'kubernetes namespace does not exist' do
+      let(:persisted_namespace) { nil }
+      let(:namespace) { 'kubernetes-namespace' }
+      let(:kubeconfig) { 'kubeconfig' }
 
-          it_behaves_like 'setting variables'
-
-          it 'sets KUBE_TOKEN' do
-            expect(subject).to include(
-              { key: 'KUBE_TOKEN', value: kubernetes.token, public: false, masked: true }
-            )
-          end
-        end
+      before do
+        allow(Gitlab::Kubernetes::DefaultNamespace).to receive(:new)
+          .with(cluster, project: project).and_return(double(from_environment_name: namespace))
+        allow(platform).to receive(:kubeconfig).with(namespace).and_return(kubeconfig)
       end
 
-      context 'kubernetes namespace exists for the project' do
-        let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token, cluster: cluster, project: project) }
+      it { is_expected.not_to include(key: 'KUBE_TOKEN', value: platform.token, public: false, masked: true) }
+      it { is_expected.not_to include(key: 'KUBE_NAMESPACE', value: namespace) }
+      it { is_expected.not_to include(key: 'KUBECONFIG', value: kubeconfig, public: false, file: true) }
 
-        it_behaves_like 'setting variables'
+      context 'cluster is unmanaged' do
+        let(:cluster) { create(:cluster, :group, :not_managed, platform_kubernetes: platform) }
 
-        it 'sets KUBE_TOKEN' do
-          expect(subject).to include(
-            { key: 'KUBE_TOKEN', value: kubernetes_namespace.service_account_token, public: false, masked: true }
-          )
-        end
+        it { is_expected.to include(key: 'KUBE_TOKEN', value: platform.token, public: false, masked: true) }
+        it { is_expected.to include(key: 'KUBE_NAMESPACE', value: namespace) }
+        it { is_expected.to include(key: 'KUBECONFIG', value: kubeconfig, public: false, file: true) }
       end
     end
 
-    context 'with a domain' do
-      let!(:cluster) do
-        create(:cluster, :provided_by_gcp, :with_domain,
-               platform_kubernetes: kubernetes)
-      end
+    context 'cluster variables' do
+      let(:variable) { Hash(key: :fake_key, value: 'fake_value') }
+      let(:cluster_variables) { Gitlab::Ci::Variables::Collection.new([variable]) }
 
-      it 'sets KUBE_INGRESS_BASE_DOMAIN' do
-        expect(subject).to include(
-          { key: 'KUBE_INGRESS_BASE_DOMAIN', value: cluster.domain, public: true }
-        )
+      before do
+        expect(cluster).to receive(:predefined_variables).and_return(cluster_variables)
       end
+
+      it { is_expected.to include(variable) }
     end
   end
 
@@ -410,7 +295,7 @@ describe Clusters::Platforms::Kubernetes do
     end
 
     context 'with valid pods' do
-      let(:pod) { kube_pod(environment_slug: environment.slug, namespace: cluster.kubernetes_namespace_for(project), project_slug: project.full_path_slug) }
+      let(:pod) { kube_pod(environment_slug: environment.slug, namespace: cluster.kubernetes_namespace_for(environment), project_slug: project.full_path_slug) }
       let(:pod_with_no_terminal) { kube_pod(environment_slug: environment.slug, project_slug: project.full_path_slug, status: "Pending") }
       let(:terminals) { kube_terminals(service, pod) }
       let(:pods) { [pod, pod, pod_with_no_terminal, kube_pod(environment_slug: "should-be-filtered-out")] }
diff --git a/spec/models/environment_spec.rb b/spec/models/environment_spec.rb
index d2e0bed721e..521c4704c87 100644
--- a/spec/models/environment_spec.rb
+++ b/spec/models/environment_spec.rb
@@ -575,6 +575,34 @@ describe Environment, :use_clean_rails_memory_store_caching do
     end
   end
 
+  describe '#deployment_namespace' do
+    let(:environment) { create(:environment) }
+
+    subject { environment.deployment_namespace }
+
+    before do
+      allow(environment).to receive(:deployment_platform).and_return(deployment_platform)
+    end
+
+    context 'no deployment platform available' do
+      let(:deployment_platform) { nil }
+
+      it { is_expected.to be_nil }
+    end
+
+    context 'deployment platform is available' do
+      let(:cluster) { create(:cluster, :provided_by_user, :project, projects: [environment.project]) }
+      let(:deployment_platform) { cluster.platform }
+
+      it 'retrieves a namespace from the cluster' do
+        expect(cluster).to receive(:kubernetes_namespace_for)
+          .with(environment).and_return('mock-namespace')
+
+        expect(subject).to eq 'mock-namespace'
+      end
+    end
+  end
+
   describe '#terminals' do
     subject { environment.terminals }
 
@@ -823,4 +851,35 @@ describe Environment, :use_clean_rails_memory_store_caching do
       subject.prometheus_adapter
     end
   end
+
+  describe '#knative_services_finder' do
+    let(:environment) { create(:environment) }
+
+    subject { environment.knative_services_finder }
+
+    context 'environment has no deployments' do
+      it { is_expected.to be_nil }
+    end
+
+    context 'environment has a deployment' do
+      let!(:deployment) { create(:deployment, :success, environment: environment, cluster: cluster) }
+
+      context 'with no cluster associated' do
+        let(:cluster) { nil }
+
+        it { is_expected.to be_nil }
+      end
+
+      context 'with a cluster associated' do
+        let(:cluster) { create(:cluster) }
+
+        it 'calls the service finder' do
+          expect(Clusters::KnativeServicesFinder).to receive(:new)
+            .with(cluster, environment).and_return(:finder)
+
+          is_expected.to eq :finder
+        end
+      end
+    end
+  end
 end
diff --git a/spec/models/project_spec.rb b/spec/models/project_spec.rb
index 2e6c81d0a7e..dde766c3813 100644
--- a/spec/models/project_spec.rb
+++ b/spec/models/project_spec.rb
@@ -2594,45 +2594,33 @@ describe Project do
   end
 
   describe '#deployment_variables' do
-    context 'when project has no deployment service' do
-      let(:project) { create(:project) }
+    let(:project) { create(:project) }
+    let(:environment) { 'production' }
 
-      it 'returns an empty array' do
-        expect(project.deployment_variables).to eq []
-      end
+    subject { project.deployment_variables(environment: environment) }
+
+    before do
+      expect(project).to receive(:deployment_platform).with(environment: environment)
+        .and_return(deployment_platform)
     end
 
-    context 'when project uses mock deployment service' do
-      let(:project) { create(:mock_deployment_project) }
+    context 'when project has no deployment platform' do
+      let(:deployment_platform) { nil }
 
-      it 'returns an empty array' do
-        expect(project.deployment_variables).to eq []
-      end
+      it { is_expected.to eq [] }
     end
 
-    context 'when project has a deployment service' do
-      context 'when user configured kubernetes from CI/CD > Clusters and KubernetesNamespace migration has not been executed' do
-        let!(:cluster) { create(:cluster, :project, :provided_by_gcp) }
-        let(:project) { cluster.project }
+    context 'when project has a deployment platform' do
+      let(:platform_variables) { %w(platform variables) }
+      let(:deployment_platform) { double }
 
-        it 'does not return variables from this service' do
-          expect(project.deployment_variables).not_to include(
-            { key: 'KUBE_TOKEN', value: project.deployment_platform.token, public: false, masked: true }
-          )
-        end
+      before do
+        expect(deployment_platform).to receive(:predefined_variables)
+          .with(project: project, environment_name: environment)
+          .and_return(platform_variables)
       end
 
-      context 'when user configured kubernetes from CI/CD > Clusters and KubernetesNamespace migration has been executed' do
-        let!(:kubernetes_namespace) { create(:cluster_kubernetes_namespace, :with_token) }
-        let!(:cluster) { kubernetes_namespace.cluster }
-        let(:project) { kubernetes_namespace.project }
-
-        it 'returns token from kubernetes namespace' do
-          expect(project.deployment_variables).to include(
-            { key: 'KUBE_TOKEN', value: kubernetes_namespace.service_account_token, public: false, masked: true }
-          )
-        end
-      end
+      it { is_expected.to eq platform_variables }
     end
   end
 
diff --git a/spec/requests/api/project_clusters_spec.rb b/spec/requests/api/project_clusters_spec.rb
index e8ed016db69..a7b919de2ef 100644
--- a/spec/requests/api/project_clusters_spec.rb
+++ b/spec/requests/api/project_clusters_spec.rb
@@ -336,7 +336,6 @@ describe API::ProjectClusters do
         it 'does not update cluster attributes' do
           expect(cluster.domain).not_to eq('new_domain.com')
           expect(cluster.platform_kubernetes.namespace).not_to eq('invalid_namespace')
-          expect(cluster.kubernetes_namespace_for(project)).not_to eq('invalid_namespace')
         end
 
         it 'returns validation errors' do
diff --git a/spec/services/clusters/build_kubernetes_namespace_service_spec.rb b/spec/services/clusters/build_kubernetes_namespace_service_spec.rb
new file mode 100644
index 00000000000..36c05469542
--- /dev/null
+++ b/spec/services/clusters/build_kubernetes_namespace_service_spec.rb
@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Clusters::BuildKubernetesNamespaceService do
+  let(:cluster) { create(:cluster, :project, :provided_by_gcp) }
+  let(:environment) { create(:environment) }
+  let(:project) { environment.project }
+
+  let(:namespace_generator) { double(from_environment_slug: namespace) }
+  let(:namespace) { 'namespace' }
+
+  subject { described_class.new(cluster, environment: environment).execute }
+
+  before do
+    allow(Gitlab::Kubernetes::DefaultNamespace).to receive(:new).and_return(namespace_generator)
+  end
+
+  shared_examples 'shared attributes' do
+    it 'initializes a new namespace and sets default values' do
+      expect(subject).to be_new_record
+      expect(subject.cluster).to eq cluster
+      expect(subject.project).to eq project
+      expect(subject.namespace).to eq namespace
+      expect(subject.service_account_name).to eq "#{namespace}-service-account"
+    end
+  end
+
+  include_examples 'shared attributes'
+
+  it 'sets cluster_project and environment' do
+    expect(subject.cluster_project).to eq cluster.cluster_project
+    expect(subject.environment).to eq environment
+  end
+
+  context 'namespace per environment is disabled' do
+    let(:cluster) { create(:cluster, :project, :provided_by_gcp, :namespace_per_environment_disabled) }
+
+    include_examples 'shared attributes'
+
+    it 'does not set environment' do
+      expect(subject.cluster_project).to eq cluster.cluster_project
+      expect(subject.environment).to be_nil
+    end
+  end
+
+  context 'group cluster' do
+    let(:cluster) { create(:cluster, :group, :provided_by_gcp) }
+
+    include_examples 'shared attributes'
+
+    it 'does not set cluster_project' do
+      expect(subject.cluster_project).to be_nil
+      expect(subject.environment).to eq environment
+    end
+  end
+end
diff --git a/spec/services/clusters/gcp/kubernetes/create_or_update_namespace_service_spec.rb b/spec/services/clusters/gcp/kubernetes/create_or_update_namespace_service_spec.rb
index 44407ae2793..e44cc3f5a78 100644
--- a/spec/services/clusters/gcp/kubernetes/create_or_update_namespace_service_spec.rb
+++ b/spec/services/clusters/gcp/kubernetes/create_or_update_namespace_service_spec.rb
@@ -9,8 +9,9 @@ describe Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService, '#execute' d
   let(:platform) { cluster.platform }
   let(:api_url) { 'https://kubernetes.example.com' }
   let(:project) { cluster.project }
+  let(:environment) { create(:environment, project: project) }
   let(:cluster_project) { cluster.cluster_project }
-  let(:namespace) { "#{project.path}-#{project.id}" }
+  let(:namespace) { "#{project.name}-#{project.id}-#{environment.slug}" }
 
   subject do
     described_class.new(
@@ -79,7 +80,8 @@ describe Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService, '#execute' d
       let(:kubernetes_namespace) do
         build(:cluster_kubernetes_namespace,
               cluster: cluster,
-              project: project)
+              project: project,
+              environment: environment)
       end
 
       it_behaves_like 'successful creation of kubernetes namespace'
@@ -92,20 +94,22 @@ describe Clusters::Gcp::Kubernetes::CreateOrUpdateNamespaceService, '#execute' d
         build(:cluster_kubernetes_namespace,
               cluster: cluster,
               project: cluster_project.project,
-              cluster_project: cluster_project)
+              cluster_project: cluster_project,
+              environment: environment)
       end
 
       it_behaves_like 'successful creation of kubernetes namespace'
     end
 
     context 'when there is a Kubernetes Namespace associated' do
-      let(:namespace) { 'new-namespace' }
+      let(:namespace) { "new-namespace-#{environment.slug}" }
 
       let(:kubernetes_namespace) do
         create(:cluster_kubernetes_namespace,
                cluster: cluster,
                project: cluster_project.project,
-               cluster_project: cluster_project)
+               cluster_project: cluster_project,
+               environment: environment)
       end
 
       before do
diff --git a/spec/support/prometheus/additional_metrics_shared_examples.rb b/spec/support/prometheus/additional_metrics_shared_examples.rb
index 82582630dee..4e006edb7da 100644
--- a/spec/support/prometheus/additional_metrics_shared_examples.rb
+++ b/spec/support/prometheus/additional_metrics_shared_examples.rb
@@ -50,7 +50,7 @@ RSpec.shared_examples 'additional metrics query' do
         let!(:cluster) { create(:cluster, :project, :provided_by_gcp) }
         let(:project) { cluster.project }
         let(:environment) { create(:environment, slug: 'environment-slug', project: project) }
-        let(:kube_namespace) { project.deployment_platform.kubernetes_namespace_for(project) }
+        let(:kube_namespace) { environment.deployment_namespace }
 
         it_behaves_like 'query context containing environment slug and filter'
 
diff --git a/spec/support/services/clusters/create_service_shared.rb b/spec/support/services/clusters/create_service_shared.rb
index 6ec8750ce87..27f6d0570b6 100644
--- a/spec/support/services/clusters/create_service_shared.rb
+++ b/spec/support/services/clusters/create_service_shared.rb
@@ -32,23 +32,56 @@ shared_context 'invalid cluster create params' do
 end
 
 shared_examples 'create cluster service success' do
-  it 'creates a cluster object and performs a worker' do
-    expect(ClusterProvisionWorker).to receive(:perform_async)
-
-    expect { subject }
-      .to change { Clusters::Cluster.count }.by(1)
-      .and change { Clusters::Providers::Gcp.count }.by(1)
-
-    expect(subject.name).to eq('test-cluster')
-    expect(subject.user).to eq(user)
-    expect(subject.project).to eq(project)
-    expect(subject.provider.gcp_project_id).to eq('gcp-project')
-    expect(subject.provider.zone).to eq('us-central1-a')
-    expect(subject.provider.num_nodes).to eq(1)
-    expect(subject.provider.machine_type).to eq('machine_type-a')
-    expect(subject.provider.access_token).to eq(access_token)
-    expect(subject.provider).to be_legacy_abac
-    expect(subject.platform).to be_nil
+  context 'namespace per environment feature is enabled' do
+    before do
+      stub_feature_flags(kubernetes_namespace_per_environment: true)
+    end
+
+    it 'creates a cluster object and performs a worker' do
+      expect(ClusterProvisionWorker).to receive(:perform_async)
+
+      expect { subject }
+        .to change { Clusters::Cluster.count }.by(1)
+        .and change { Clusters::Providers::Gcp.count }.by(1)
+
+      expect(subject.name).to eq('test-cluster')
+      expect(subject.user).to eq(user)
+      expect(subject.project).to eq(project)
+      expect(subject.provider.gcp_project_id).to eq('gcp-project')
+      expect(subject.provider.zone).to eq('us-central1-a')
+      expect(subject.provider.num_nodes).to eq(1)
+      expect(subject.provider.machine_type).to eq('machine_type-a')
+      expect(subject.provider.access_token).to eq(access_token)
+      expect(subject.provider).to be_legacy_abac
+      expect(subject.platform).to be_nil
+      expect(subject.namespace_per_environment).to eq true
+    end
+  end
+
+  context 'namespace per environment feature is disabled' do
+    before do
+      stub_feature_flags(kubernetes_namespace_per_environment: false)
+    end
+
+    it 'creates a cluster object and performs a worker' do
+      expect(ClusterProvisionWorker).to receive(:perform_async)
+
+      expect { subject }
+        .to change { Clusters::Cluster.count }.by(1)
+        .and change { Clusters::Providers::Gcp.count }.by(1)
+
+      expect(subject.name).to eq('test-cluster')
+      expect(subject.user).to eq(user)
+      expect(subject.project).to eq(project)
+      expect(subject.provider.gcp_project_id).to eq('gcp-project')
+      expect(subject.provider.zone).to eq('us-central1-a')
+      expect(subject.provider.num_nodes).to eq(1)
+      expect(subject.provider.machine_type).to eq('machine_type-a')
+      expect(subject.provider.access_token).to eq(access_token)
+      expect(subject.provider).to be_legacy_abac
+      expect(subject.platform).to be_nil
+      expect(subject.namespace_per_environment).to eq false
+    end
   end
 end