diff --git a/app/models/container_label_tag_mapping.rb b/app/models/container_label_tag_mapping.rb
index f7392593c7e..655b64a875c 100644
--- a/app/models/container_label_tag_mapping.rb
+++ b/app/models/container_label_tag_mapping.rb
@@ -9,91 +9,93 @@ class ContainerLabelTagMapping < ApplicationRecord
   # - When `label_value` is NULL, we map this name with any value to per-value tags.
   #   In this case, `tag` specifies the category under which to create
   #   the value-specific tag (and classification) on demand.
-  #   We then also add a specific `label_value`->specific `tag` mapping here.
+  #
+  # All involved tags must also have a Classification.
 
   belongs_to :tag
 
-  def self.drop_cache
-    @hash_all_by_name_type_value = nil
-  end
-
-  # Returns {[name, type, value] => [tag, ...]}}} hash.
-  def self.hash_all_by_name_type_value
-    unless @hash_all_by_name_type_value
-      @hash_all_by_name_type_value = {}
-      includes(:tag).find_each { |m| load_mapping_into_hash(m) }
-    end
-    @hash_all_by_name_type_value
-  end
-
-  def self.load_mapping_into_hash(mapping)
-    return unless @hash_all_by_name_type_value
-    key = [mapping.label_name, mapping.labeled_resource_type, mapping.label_value].freeze
-    @hash_all_by_name_type_value[key] ||= []
-    @hash_all_by_name_type_value[key] << mapping.tag
+  # Pass the data this returns to map_* methods.
+  def self.cache
+    # {[name, type, value] => [tag_id, ...]}
+    in_my_region.find_each
+                .group_by { |m| [m.label_name, m.labeled_resource_type, m.label_value].freeze }
+                .transform_values { |mappings| mappings.collect(&:tag_id) }
   end
-  private_class_method :load_mapping_into_hash
 
-  # All specific-value tags that can be assigned by this mapping.
-  def self.mappable_tags
-    hash_all_by_name_type_value.collect_concat do |(_name, _type, value), tags|
-      value ? tags : []
-    end
-  end
+  # We expect labels to be {:name, :value} hashes
+  # and return {:tag_id} or {:category_tag_id, :entry_name, :entry_description} hashes.
 
-  # Main entry point.
-  def self.tags_for_entity(entity)
-    entity.labels.collect_concat { |label| tags_for_label(label) }
+  def self.map_labels(cache, type, labels)
+    labels.collect_concat { |label| map_label(cache, type, label) }.uniq
   end
 
-  def self.tags_for_label(label)
+  def self.map_label(cache, type, label)
     # Apply both specific-type and any-type, independently.
-    (tags_for_name_type_value(label.name, label.resource_type, label.value) +
-     tags_for_name_type_value(label.name, nil,                 label.value))
+    (map_name_type_value(cache, label[:name], type, label[:value]) +
+     map_name_type_value(cache, label[:name], nil,  label[:value]))
   end
 
-  def self.tags_for_name_type_value(name, type, value)
-    specific_value = hash_all_by_name_type_value[[name, type, value]] || []
-    any_value      = hash_all_by_name_type_value[[name, type, nil]]   || []
+  def self.map_name_type_value(cache, name, type, value)
+    specific_value = cache[[name, type, value]] || []
+    any_value      = cache[[name, type, nil]]   || []
     if !specific_value.empty?
-      specific_value
+      specific_value.map { |tag_id| {:tag_id => tag_id} }
     else
-      any_value.map do |category_tag|
-        create_specific_value_mapping(name, type, value, category_tag).tag
+      if value.empty?
+        [] # Don't map empty value to any tag.
+      else
+        # Note: if the way we compute `entry_name` changes,
+        # consider what will happen to previously created tags.
+        any_value.map do |tag_id|
+          {:category_tag_id   => tag_id,
+           :entry_name        => Classification.sanitize_name(value),
+           :entry_description => value}
+        end
       end
     end
   end
-  private_class_method :tags_for_name_type_value
+  private_class_method :map_name_type_value
 
-  # If this is an open ended any-value mapping, finds or creates a
-  # specific-value mapping to a specific tag.
-  def self.create_specific_value_mapping(name, type, value, category_tag)
-    new_tag = create_tag(name, value, category_tag)
-    new_mapping = create!(:labeled_resource_type => type, :label_name => name, :label_value => value,
-                          :tag => new_tag)
-    load_mapping_into_hash(new_mapping)
-    new_mapping
+  # Given a hash built by `map_*` methods, returns a Tag (creating if needed).
+  def self.find_or_create_tag(tag_hash)
+    if tag_hash[:tag_id]
+      Tag.find(tag_hash[:tag_id])
+    else
+      category = Tag.find(tag_hash[:category_tag_id]).classification
+      entry = category.find_entry_by_name(tag_hash[:entry_name])
+      unless entry
+        category.lock :exclusive do
+          begin
+            entry = category.add_entry(:name        => tag_hash[:entry_name],
+                                       :description => tag_hash[:entry_description])
+            entry.save!
+          rescue ActiveRecord::RecordInvalid
+            entry = category.find_entry_by_name(tag_hash[:entry_name])
+          end
+        end
+      end
+      entry.tag
+    end
   end
-  private_class_method :create_specific_value_mapping
 
-  def self.create_tag(name, value, category_tag)
-    category = category_tag.classification
-    unless category
-      category = Classification.create_category!(:description => "Kubernetes label '#{name}'",
-                                                 :read_only   => true,
-                                                 :tag         => category_tag)
-    end
+  def self.controls_tag?(tag)
+    return false unless tag.classification.try(:read_only) # never touch user-assignable tags.
+    tag_ids = [tag.id, tag.category.tag_id].uniq
+    where(:tag_id => tag_ids).any?
+  end
 
-    if value.empty?
-      entry_name = ':empty:' # ':' character won't occur in kubernetes values.
-      description = '<empty value>'
-    else
-      entry_name = Classification.sanitize_name(value)
-      description = value
+  # Assign/unassign mapping-controlled tags, preserving user-assigned tags.
+  def self.retag_entity(entity, tag_hashes)
+    mapped_tags = tag_hashes.map { |tag_hash| find_or_create_tag(tag_hash) }
+    existing_tags = entity.tags
+    Tagging.transaction do
+      (mapped_tags - existing_tags).each do |tag|
+        Tagging.create!(:taggable => entity, :tag => tag)
+      end
+      (existing_tags - mapped_tags).select { |tag| controls_tag?(tag) }.tap do |tags|
+        Tagging.where(:taggable => entity, :tag => tags.collect(&:id)).destroy_all
+      end
     end
-    entry = category.add_entry(:name => entry_name, :description => description)
-    entry.save!
-    entry.tag
+    entity.tags.reset
   end
-  private_class_method :create_tag
 end
diff --git a/app/models/ems_refresh/save_inventory_container.rb b/app/models/ems_refresh/save_inventory_container.rb
index e1feb6a0e0f..de6ab0afdec 100644
--- a/app/models/ems_refresh/save_inventory_container.rb
+++ b/app/models/ems_refresh/save_inventory_container.rb
@@ -28,7 +28,7 @@ def save_container_projects_inventory(ems, hashes, target = nil)
               end
 
     save_inventory_multi(ems.container_projects, hashes, deletes, [:ems_ref],
-                         [:labels_and_tags], [], true)
+                         [:labels, :tags], [], true)
     store_ids_for_new_records(ems.container_projects, hashes, :ems_ref)
   end
 
@@ -146,7 +146,7 @@ def save_container_routes_inventory(ems, hashes, target = nil)
     end
 
     save_inventory_multi(ems.container_routes, hashes, deletes, [:ems_ref],
-                         [:labels_and_tags], [:container_service, :project, :namespace])
+                         [:labels, :tags], [:container_service, :project, :namespace])
     store_ids_for_new_records(ems.container_routes, hashes, :ems_ref)
   end
 
@@ -162,7 +162,7 @@ def save_container_nodes_inventory(ems, hashes, target = nil)
               end
 
     save_inventory_multi(ems.container_nodes, hashes, deletes, [:ems_ref],
-                         [:labels_and_tags, :computer_system, :container_conditions], [:namespace])
+                         [:labels, :tags, :computer_system, :container_conditions], [:namespace])
     store_ids_for_new_records(ems.container_nodes, hashes, :ems_ref)
   end
 
@@ -186,7 +186,7 @@ def save_container_replicators_inventory(ems, hashes, target = nil)
     end
 
     save_inventory_multi(ems.container_replicators, hashes, deletes, [:ems_ref],
-                         [:labels_and_tags, :selector_parts], [:project, :namespace])
+                         [:labels, :tags, :selector_parts], [:project, :namespace])
     store_ids_for_new_records(ems.container_replicators, hashes, :ems_ref)
   end
 
@@ -208,7 +208,7 @@ def save_container_services_inventory(ems, hashes, target = nil)
     end
 
     save_inventory_multi(ems.container_services, hashes, deletes, [:ems_ref],
-                         [:labels_and_tags, :selector_parts, :container_service_port_configs],
+                         [:labels, :tags, :selector_parts, :container_service_port_configs],
                          [:container_groups, :project, :container_image_registry, :namespace])
 
     store_ids_for_new_records(ems.container_services, hashes, :ems_ref)
@@ -234,7 +234,7 @@ def save_container_groups_inventory(ems, hashes, target = nil)
     end
 
     save_inventory_multi(ems.container_groups, hashes, deletes, [:ems_ref],
-                         [:container_definitions, :containers, :labels_and_tags,
+                         [:container_definitions, :containers, :labels, :tags,
                           :node_selector_parts, :container_conditions, :container_volumes],
                          [:container_node, :container_replicator, :project, :namespace, :build_pod_name],
                          true)
@@ -409,21 +409,14 @@ def save_labels_inventory(entity, hashes, target = nil)
     store_ids_for_new_records(entity.labels, hashes, [:section, :name])
   end
 
-  def save_labels_and_tags_inventory(entity, hashes, target = nil)
-    save_labels_inventory(entity, hashes, target)
-    save_tags_generated_from_labels(entity, hashes, target)
-  end
-
-  def save_tags_generated_from_labels(entity, hashes, _target = nil)
-    labels = hashes.collect do |label_hash|
-      OpenStruct.new(:resource_type => entity.class.base_class.name,
-                     :name          => label_hash[:name],
-                     :value         => label_hash[:value])
-    end
-    current_tags = labels.collect_concat { |label| ContainerLabelTagMapping.tags_for_label(label) }
-    mappable_tags = ContainerLabelTagMapping.mappable_tags
+  def save_tags_inventory(entity, hashes, _target = nil)
+    return if hashes.nil?
 
-    entity.tags = entity.tags - mappable_tags + current_tags
+    ContainerLabelTagMapping.retag_entity(entity, hashes) # Keeps user-assigned tags.
+  rescue => err
+    raise if EmsRefresh.debug_failures
+    _log.error("Auto-tagging failed on #{entity.class} [#{entity.name}] with error [#{err}].")
+    _log.log_backtrace(err)
   end
 
   def save_selector_parts_inventory(entity, hashes, target = nil)
@@ -464,7 +457,7 @@ def save_container_builds_inventory(ems, hashes, target = nil)
     hashes.each do |h|
       h[:container_project_id] = h.fetch_path(:project, :id)
     end
-    save_inventory_multi(ems.container_builds, hashes, deletes, [:ems_ref], [:labels_and_tags],
+    save_inventory_multi(ems.container_builds, hashes, deletes, [:ems_ref], [:labels, :tags],
                          [:project, :resources])
     store_ids_for_new_records(ems.container_builds, hashes, :ems_ref)
   end
diff --git a/app/models/manageiq/providers/kubernetes/container_manager/refresh_parser.rb b/app/models/manageiq/providers/kubernetes/container_manager/refresh_parser.rb
index c1c45b8d19e..f574cac2b65 100644
--- a/app/models/manageiq/providers/kubernetes/container_manager/refresh_parser.rb
+++ b/app/models/manageiq/providers/kubernetes/container_manager/refresh_parser.rb
@@ -10,6 +10,7 @@ def self.ems_inv_to_hashes(inventory)
     def initialize
       @data = {}
       @data_index = {}
+      @label_tag_mapping = ContainerLabelTagMapping.cache
     end
 
     def ems_inv_to_hashes(inventory)
@@ -138,6 +139,10 @@ def process_collection_item(item, key)
       new_result
     end
 
+    def map_labels(model_name, labels)
+      ContainerLabelTagMapping.map_labels(@label_tag_mapping, model_name, labels)
+    end
+
     def find_host_by_provider_id(provider_id)
       scheme, instance_uri = provider_id.split("://", 2)
       prov, name_field = scheme_to_provider_mapping[scheme]
@@ -187,12 +192,14 @@ def cross_link_node(new_result)
     def parse_node(node)
       new_result = parse_base_item(node)
 
+      labels = parse_labels(node)
       new_result.merge!(
-        :type            => 'ManageIQ::Providers::Kubernetes::ContainerManager::ContainerNode',
-        :identity_infra  => node.spec.providerID,
-        :labels_and_tags => parse_labels(node),
-        :lives_on_id     => nil,
-        :lives_on_type   => nil
+        :type           => 'ManageIQ::Providers::Kubernetes::ContainerManager::ContainerNode',
+        :identity_infra => node.spec.providerID,
+        :labels         => labels,
+        :tags           => map_labels('ContainerNode', labels),
+        :lives_on_id    => nil,
+        :lives_on_type  => nil
       )
 
       node_info = node.status.try(:nodeInfo)
@@ -250,13 +257,14 @@ def parse_service(service)
         container_groups << cg unless cg.nil?
       end
 
+      labels = parse_labels(service)
       new_result.merge!(
         # TODO: We might want to change portal_ip to clusterIP
         :portal_ip        => service.spec.clusterIP,
         :session_affinity => service.spec.sessionAffinity,
         :service_type     => service.spec.type,
-
-        :labels_and_tags  => parse_labels(service),
+        :labels           => labels,
+        :tags             => map_labels('ContainerService', labels),
         :selector_parts   => parse_selector_parts(service),
         :container_groups => container_groups
       )
@@ -333,7 +341,8 @@ def parse_pod(pod)
 
       new_result[:container_conditions] = parse_conditions(pod)
 
-      new_result[:labels_and_tags] = parse_labels(pod)
+      new_result[:labels] = parse_labels(pod)
+      new_result[:tags] = map_labels('ContainerGroup', new_result[:labels])
       new_result[:node_selector_parts] = parse_node_selector_parts(pod)
       new_result[:container_volumes] = parse_volumes(pod)
       new_result
@@ -360,7 +369,8 @@ def parse_endpoint(entity)
 
     def parse_namespace(namespace)
       new_result = parse_base_item(namespace).except(:namespace)
-      new_result[:labels_and_tags] = parse_labels(namespace)
+      new_result[:labels] = parse_labels(namespace)
+      new_result[:tags] = map_labels('ContainerProject', new_result[:labels])
       new_result
     end
 
@@ -511,11 +521,13 @@ def create_limits_matrix
     def parse_replication_controllers(container_replicator)
       new_result = parse_base_item(container_replicator)
 
+      labels = parse_labels(container_replicator)
       # TODO: parse template
       new_result.merge!(
         :replicas         => container_replicator.spec.replicas,
         :current_replicas => container_replicator.status.replicas,
-        :labels_and_tags  => parse_labels(container_replicator),
+        :labels           => labels,
+        :tags             => map_labels('ContainerReplicator', labels),
         :selector_parts   => parse_selector_parts(container_replicator)
       )
 
diff --git a/app/models/manageiq/providers/openshift/container_manager/refresh_parser.rb b/app/models/manageiq/providers/openshift/container_manager/refresh_parser.rb
index c13b491278b..47f6259e6f6 100644
--- a/app/models/manageiq/providers/openshift/container_manager/refresh_parser.rb
+++ b/app/models/manageiq/providers/openshift/container_manager/refresh_parser.rb
@@ -61,11 +61,13 @@ def get_service_name(route)
       def parse_route(route)
         new_result = parse_base_item(route)
 
+        labels = parse_labels(route)
         new_result.merge!(
           # TODO: persist tls
-          :host_name       => route.spec.try(:host),
-          :labels_and_tags => parse_labels(route),
-          :path            => route.path
+          :host_name => route.spec.try(:host),
+          :labels    => labels,
+          :tags      => map_labels('ContainerRoute', labels),
+          :path      => route.path
         )
 
         new_result[:project] = @data_index.fetch_path(:container_projects, :by_name,
@@ -89,8 +91,10 @@ def parse_build_source(source_item)
       def parse_build(build)
         new_result = parse_base_item(build)
         new_result.merge! parse_build_source(build.spec.source)
+        labels = parse_labels(build)
         new_result.merge!(
-          :labels_and_tags             => parse_labels(build),
+          :labels                      => labels,
+          :tags                        => map_labels('ContainerBuild', labels),
           :service_account             => build.spec.serviceAccount,
           :completion_deadline_seconds => build.spec.try(:completionDeadlineSeconds),
           :output_name                 => build.spec.try(:output).try(:to).try(:name)
diff --git a/spec/models/container_label_tag_mapping_spec.rb b/spec/models/container_label_tag_mapping_spec.rb
index e6ea3e4deb9..4b3c5c21ef6 100644
--- a/spec/models/container_label_tag_mapping_spec.rb
+++ b/spec/models/container_label_tag_mapping_spec.rb
@@ -1,123 +1,194 @@
 describe ContainerLabelTagMapping do
-  let(:node) { FactoryGirl.create(:container_node, :name => 'node') }
-  let(:project) { FactoryGirl.create(:container_project, :name => 'project') }
-
-  def label(node, name, value)
-    FactoryGirl.create(:kubernetes_label, :resource => node, :name => name, :value => value)
-  end
-
-  let(:tag1) { FactoryGirl.create(:tag, :name => '/ns1/category1/tag1') }
-  let(:tag2) { FactoryGirl.create(:tag, :name => '/ns2/category2/tag2') }
-  let(:cat_tag_without_classification) { FactoryGirl.create(:tag, :name => '/ns3/category3') }
-
-  let(:cat_classification) { FactoryGirl.create(:classification) }
+  let(:cat_classification) { FactoryGirl.create(:classification, :read_only => true) }
   let(:cat_tag) { cat_classification.tag }
-  let(:tag_under_cat) { cat_classification.add_entry(:name => 'value_1', :description => 'value-1').tag }
+  let(:tag1) { cat_classification.add_entry(:name => 'value_1', :description => 'value-1').tag }
+  let(:tag2) { cat_classification.add_entry(:name => 'something_else', :description => 'Another tag').tag }
+  let(:tag3) { cat_classification.add_entry(:name => 'yet_another', :description => 'Yet another tag').tag }
   let(:empty_tag_under_cat) do
     cat_classification.add_entry(:name => 'my_empty', :description => 'Custom description for empty value').tag
   end
+  let(:tag_in_another_cat) do
+    cat = FactoryGirl.create(:classification, :read_only => true)
+    cat.add_entry(:name => 'unrelated', :description => 'Unrelated tag').tag
+  end
+
+  def labels(kv)
+    kv.map do |name, value|
+      {:section => 'labels', :source => 'kubernetes',
+       :name => name, :value => value}
+    end
+  end
+
+  def map_labels(model_name, labels_kv)
+    ContainerLabelTagMapping.map_labels(ContainerLabelTagMapping.cache,
+                                        model_name,
+                                        labels(labels_kv))
+  end
 
-  # Each test here may populate the table differently.
-  after :each do
-    ContainerLabelTagMapping.drop_cache
+  def to_tags(tag_hashes)
+    tag_hashes.map { |h| ContainerLabelTagMapping.find_or_create_tag(h) }
   end
-  # If the mapping was called from *elsewhere* there might already be a stale cache.
-  # TODO: This assumes all other tests only use an empty mapping.
-  before :all do
-    ContainerLabelTagMapping.drop_cache
+
+  # All-in-one
+  def map_to_tags(model_name, labels_kv)
+    to_tags(map_labels(model_name, labels_kv))
   end
 
   context "with empty mapping" do
     it "does nothing" do
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to be_empty
-      expect(ContainerLabelTagMapping.mappable_tags).to be_empty
+      expect(ContainerLabelTagMapping.cache).to be_empty
+      expect(map_labels('ContainerNode',
+                        'foo'  => 'bar',
+                        'quux' => 'whatever')).to be_empty
     end
   end
 
   context "with 2 mappings for same label" do
-    before do
+    before(:each) do
       FactoryGirl.create(:container_label_tag_mapping, :only_nodes, :label_value => 'value-1', :tag => tag1)
       FactoryGirl.create(:container_label_tag_mapping, :only_nodes, :label_value => 'value-1', :tag => tag2)
     end
 
-    it "tags_for_entity returns 2 tags" do
-      label(node, 'name', 'value-1')
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to contain_exactly(tag1, tag2)
-    end
-
-    it "tags_for_label returns same tags" do
-      label_obj = OpenStruct.new(:resource_type => 'ContainerNode',
-                                 :name          => 'name',
-                                 :value         => 'value-1')
-      expect(ContainerLabelTagMapping.tags_for_label(label_obj)).to contain_exactly(tag1, tag2)
+    it "map_labels returns 2 tags" do
+      expect(map_labels('ContainerNode', 'name' => 'value-1')).to contain_exactly(
+        {:tag_id => tag1.id},
+        {:tag_id => tag2.id}
+      )
+      expect(map_to_tags('ContainerNode', 'name' => 'value-1')).to contain_exactly(tag1, tag2)
     end
   end
 
   context "with any-value and specific-value mappings" do
-    before do
+    before(:each) do
       FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag)
       FactoryGirl.create(:container_label_tag_mapping, :label_value => 'value-1', :tag => tag1)
-      FactoryGirl.create(:container_label_tag_mapping, :label_value => 'value-1', :tag => tag_under_cat)
-      # Force a tag to exist that we don't map to (for testing .mappable_tags).
-      tag2
+      FactoryGirl.create(:container_label_tag_mapping, :label_value => 'value-1', :tag => tag2)
+      # Force a tag to exist that we don't map to (for testing .controls_tag?).
+      tag_in_another_cat
     end
 
     it "prefers specific-value" do
-      label(node, 'name', 'value-1')
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to contain_exactly(tag1, tag_under_cat)
+      expect(map_to_tags('ContainerNode', 'name' => 'value-1')).to contain_exactly(tag1, tag2)
     end
 
     it "creates tag for new value" do
-      expect(ContainerLabelTagMapping.mappable_tags).to contain_exactly(tag1, tag_under_cat)
+      expect(ContainerLabelTagMapping.controls_tag?(tag1)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag2)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag_in_another_cat)).to be false
 
-      label(node, 'name', 'value-2')
-      tags = ContainerLabelTagMapping.tags_for_entity(node)
+      cached1 = ContainerLabelTagMapping.cache
+      tags = to_tags(ContainerLabelTagMapping.map_labels(cached1, 'ContainerNode', labels('name' => 'value-2')))
       expect(tags.size).to eq(1)
       expect(tags[0].name).to eq(cat_tag.name + '/value_2')
       expect(tags[0].classification.description).to eq('value-2')
 
-      expect(ContainerLabelTagMapping.mappable_tags).to contain_exactly(tag1, tag_under_cat, tags[0])
+      expect(ContainerLabelTagMapping.controls_tag?(tag1)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag2)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tags[0])).to be true
 
       # But nothing changes when called again, the previously created tag is re-used.
 
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to contain_exactly(tags[0])
+      tags2 = to_tags(ContainerLabelTagMapping.map_labels(cached1, 'ContainerNode', labels('name' => 'value-2')))
+      expect(tags2).to contain_exactly(tags[0])
+
+      expect(ContainerLabelTagMapping.controls_tag?(tag1)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag2)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tags[0])).to be true
+
+      # And nothing changes when we re-load the mappings table.
 
-      expect(ContainerLabelTagMapping.mappable_tags).to contain_exactly(tag1, tag_under_cat, tags[0])
+      cached2 = ContainerLabelTagMapping.cache
+
+      tags2 = to_tags(ContainerLabelTagMapping.map_labels(cached2, 'ContainerNode', labels('name' => 'value-2')))
+      expect(tags2).to contain_exactly(tags[0])
+
+      expect(ContainerLabelTagMapping.controls_tag?(tag1)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag2)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tags[0])).to be true
     end
-  end
 
-  context "given an empty label value" do
-    before do
-      label(node, 'name', '')
+    it "handles names that differ only by case" do
+      # Kubernetes names are case-sensitive
+      # (but the optional domain prefix must be lowercase).
+      FactoryGirl.create(:container_label_tag_mapping,
+                         :label_name => 'Name_Case', :label_value => 'value', :tag => tag2)
+      tags = map_to_tags('ContainerNode', 'name_case' => 'value')
+      tags2 = map_to_tags('ContainerNode', 'Name_Case' => 'value', 'naME_caSE' => 'value')
+      expect(tags).to be_empty
+      expect(tags2).to contain_exactly(tag2)
+
+      # Note: this test doesn't cover creation of the category, eg. you can't have
+      # /managed/kubernetes:name vs /managed/kubernetes:naME.
     end
 
-    it "any-value mapping generates a tag" do
-      FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag)
-      tags = ContainerLabelTagMapping.tags_for_entity(node)
-      expect(tags.size).to eq(1)
-      expect(tags[0].classification.description).to eq('<empty value>')
+    it "handles values that differ only by case / punctuation" do
+      tags = map_to_tags('ContainerNode', 'name' => 'value-case.punct')
+      tags2 = map_to_tags('ContainerNode', 'name' => 'VaLuE-CASE.punct')
+      tags3 = map_to_tags('ContainerNode', 'name' => 'value-case/punct')
+      # TODO: They get mapped to the same tag, is this desired?
+      # TODO: What do we want the description to be?
+      expect(tags2).to eq(tags)
+      expect(tags3).to eq(tags)
     end
 
-    it "honors specific mapping for the empty value" do
-      FactoryGirl.create(:container_label_tag_mapping, :label_value => '', :tag => empty_tag_under_cat)
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to contain_exactly(empty_tag_under_cat)
-      # same with both any-value and specific-value mappings
-      FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag)
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to contain_exactly(empty_tag_under_cat)
+    it "handles values that differ only past 50th character" do
+      tags = map_to_tags('ContainerNode', 'name' => 'x' * 50)
+      tags2 = map_to_tags('ContainerNode', 'name' => 'x' * 50 + 'y')
+      tags3 = map_to_tags('ContainerNode', 'name' => 'x' * 50 + 'z')
+      # TODO: They get mapped to the same tag, is this desired?
+      # TODO: What do we want the description to be?
+      expect(tags2).to eq(tags)
+      expect(tags3).to eq(tags)
+    end
+
+    # With multiple providers you'll have multiple refresh workers,
+    # each with independently cached mapping table.
+    context "2 workers with independent cache" do
+      it "handle known value simultaneously" do
+        cached1 = ContainerLabelTagMapping.cache
+        cached2 = ContainerLabelTagMapping.cache
+        tags1 = to_tags(ContainerLabelTagMapping.map_labels(cached1, 'ContainerNode', labels('name' => 'value-1')))
+        tags2 = to_tags(ContainerLabelTagMapping.map_labels(cached2, 'ContainerNode', labels('name' => 'value-1')))
+        expect(tags1).to contain_exactly(tag1, tag2)
+        expect(tags2).to contain_exactly(tag1, tag2)
+      end
+
+      it "handle new value encountered simultaneously" do
+        cached1 = ContainerLabelTagMapping.cache
+        cached2 = ContainerLabelTagMapping.cache
+        tags1 = to_tags(ContainerLabelTagMapping.map_labels(cached1, 'ContainerNode', labels('name' => 'value-2')))
+        tags2 = to_tags(ContainerLabelTagMapping.map_labels(cached2, 'ContainerNode', labels('name' => 'value-2')))
+        expect(tags1.size).to eq(1)
+        expect(tags1).to eq(tags2)
+      end
     end
   end
 
-  context "with any-value mapping whose tag has no classification" do
-    before do
-      FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag_without_classification)
+  context "with 2 any-value mappings onto same category" do
+    before(:each) do
+      FactoryGirl.create(:container_label_tag_mapping, :label_name => 'name1', :tag => cat_tag)
+      FactoryGirl.create(:container_label_tag_mapping, :label_name => 'name2', :tag => cat_tag)
     end
 
-    it "creates specific-value tag and the 2 needed classifications" do
-      label(node, 'name', 'value-3')
-      tags = ContainerLabelTagMapping.tags_for_entity(node)
+    it "maps same new value in both into 1 new tag" do
+      tags = map_to_tags('ContainerNode', 'name1' => 'value', 'name2' => 'value')
       expect(tags.size).to eq(1)
-      expect(tags[0].category.description).to eq("Kubernetes label 'name'")
-      expect(tags[0].classification.description).to eq('value-3')
+      expect(ContainerLabelTagMapping.controls_tag?(tags[0])).to be true
+    end
+  end
+
+  context "given a label with empty value" do
+    it "any-value mapping is ignored" do
+      FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag)
+      expect(map_to_tags('ContainerNode', 'name' => '')).to be_empty
+    end
+
+    it "honors specific mapping for the empty value" do
+      FactoryGirl.create(:container_label_tag_mapping, :label_value => '', :tag => empty_tag_under_cat)
+      expect(map_to_tags('ContainerNode', 'name' => '')).to contain_exactly(empty_tag_under_cat)
+      # same with both any-value and specific-value mappings
+      FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag)
+      expect(map_to_tags('ContainerNode', 'name' => '')).to contain_exactly(empty_tag_under_cat)
     end
   end
 
@@ -126,33 +197,63 @@ def label(node, name, value)
   # seemed the simplest well-defined behavior...
 
   context "with any-type and specific-type mappings" do
-    before do
+    before(:each) do
       FactoryGirl.create(:container_label_tag_mapping, :only_nodes, :label_value => 'value', :tag => tag1)
       FactoryGirl.create(:container_label_tag_mapping, :label_value => 'value', :tag => tag2)
     end
 
     it "applies both independently" do
-      label(node, 'name', 'value')
-      expect(ContainerLabelTagMapping.tags_for_entity(node)).to contain_exactly(tag1, tag2)
+      expect(map_to_tags('ContainerNode', 'name' => 'value')).to contain_exactly(tag1, tag2)
     end
 
     it "skips specific-type when type doesn't match" do
-      label(project, 'name', 'value')
-      expect(ContainerLabelTagMapping.tags_for_entity(project)).to contain_exactly(tag2)
+      expect(map_to_tags('ContainerProject', 'name' => 'value')).to contain_exactly(tag2)
     end
   end
 
   context "any-type specific-value vs specific-type any-value" do
-    before do
+    before(:each) do
       FactoryGirl.create(:container_label_tag_mapping, :only_nodes, :tag => cat_tag)
       FactoryGirl.create(:container_label_tag_mapping, :label_value => 'value', :tag => tag2)
     end
 
     it "resolves them independently" do
-      label(node, 'name', 'value')
-      tags = ContainerLabelTagMapping.tags_for_entity(node)
+      tags = map_to_tags('ContainerNode', 'name' => 'value')
       expect(tags.size).to eq(2)
       expect(tags).to include(tag2)
     end
   end
+
+  describe ".retag_entity" do
+    let(:node) { FactoryGirl.create(:container_node) }
+    before(:each) do
+      # For tag1, tag2 etc. to be considered controlled by the mapping
+      FactoryGirl.create(:container_label_tag_mapping, :tag => cat_tag)
+    end
+
+    it "assigns new tags, idempotently" do
+      expect(node.tags).to be_empty
+      ContainerLabelTagMapping.retag_entity(node, [{:tag_id => tag1.id}])
+      expect(node.tags).to contain_exactly(tag1)
+      ContainerLabelTagMapping.retag_entity(node, [{:tag_id => tag1.id}])
+      expect(node.tags).to contain_exactly(tag1)
+    end
+
+    it "unassigns obsolete mapping-controlled tags" do
+      node.tags = [tag1]
+      ContainerLabelTagMapping.retag_entity(node, [])
+      expect(node.tags).to be_empty
+    end
+
+    it "preserves user tags" do
+      user_tag = FactoryGirl.create(:tag, :name => '/managed/mycat/mytag')
+      expect(ContainerLabelTagMapping.controls_tag?(user_tag)).to be false
+      expect(ContainerLabelTagMapping.controls_tag?(tag1)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag2)).to be true
+      expect(ContainerLabelTagMapping.controls_tag?(tag3)).to be true
+      node.tags = [tag1, user_tag, tag2]
+      ContainerLabelTagMapping.retag_entity(node, [{:tag_id => tag1.id}, {:tag_id => tag3.id}])
+      expect(node.tags).to contain_exactly(user_tag, tag1, tag3)
+    end
+  end
 end
diff --git a/spec/models/manageiq/providers/kubernetes/container_manager/refresh_parser_spec.rb b/spec/models/manageiq/providers/kubernetes/container_manager/refresh_parser_spec.rb
index b6d73baadfb..07bbcb4e3b7 100644
--- a/spec/models/manageiq/providers/kubernetes/container_manager/refresh_parser_spec.rb
+++ b/spec/models/manageiq/providers/kubernetes/container_manager/refresh_parser_spec.rb
@@ -24,14 +24,15 @@
                                  :name             => "proj2",
                                  :ems_created_on   => "2016-03-28T15:04:13Z",
                                  :resource_version => "150569",
-                                 :labels_and_tags  => [
+                                 :labels           => [
                                    {
                                      :section => "labels",
                                      :name    => "department",
                                      :value   => "Warp-drive",
                                      :source  => "kubernetes"
                                    }
-                                 ])
+                                 ],
+                                 :tags             => [])
     end
   end
 
@@ -740,7 +741,8 @@
         :identity_system            => 'uuid',
         :kubernetes_kubelet_version => nil,
         :kubernetes_proxy_version   => nil,
-        :labels_and_tags            => [],
+        :labels                     => [],
+        :tags                       => [],
         :lives_on_id                => nil,
         :lives_on_type              => nil,
         :max_container_groups       => nil,
@@ -792,7 +794,8 @@
         :identity_system            => 'uuid',
         :kubernetes_kubelet_version => nil,
         :kubernetes_proxy_version   => nil,
-        :labels_and_tags            => [],
+        :labels                     => [],
+        :tags                       => [],
         :lives_on_id                => nil,
         :lives_on_type              => nil,
         :max_container_groups       => nil,
@@ -836,7 +839,8 @@
           :ems_created_on       => '2016-01-01T11:10:21Z',
           :container_conditions => [],
           :identity_infra       => 'aws:///zone/aws-id',
-          :labels_and_tags      => [],
+          :labels               => [],
+          :tags                 => [],
           :lives_on_id          => nil,
           :lives_on_type        => nil,
           :max_container_groups => nil,
diff --git a/spec/models/manageiq/providers/kubernetes/container_manager/refresher_spec.rb b/spec/models/manageiq/providers/kubernetes/container_manager/refresher_spec.rb
index 1b62ae68b30..177af532f2f 100644
--- a/spec/models/manageiq/providers/kubernetes/container_manager/refresher_spec.rb
+++ b/spec/models/manageiq/providers/kubernetes/container_manager/refresher_spec.rb
@@ -19,6 +19,15 @@
     expect(described_class.ems_type).to eq(:kubernetes)
   end
 
+  # Smoke test the use of ContainerLabelTagMapping during refresh.
+  before :each do
+    @name_category = FactoryGirl.create(:classification, :name => 'name', :description => 'Name')
+    @label_tag_mapping = FactoryGirl.create(
+      :container_label_tag_mapping,
+      :label_name => 'name', :tag => @name_category.tag
+    )
+  end
+
   it "will perform a full refresh on k8s" do
     2.times do # Run twice to verify that a second run with existing data does not change anything
       VCR.use_cassette(described_class.name.underscore) do # , :record => :new_episodes) do
@@ -138,7 +147,12 @@ def assert_specific_container_group
       :dns_policy     => "ClusterFirst",
       :phase          => "Running",
     )
-    expect(@containergroup.labels.count).to eq(1)
+    expect(@containergroup.labels).to contain_exactly(
+      label_with_name_value("name", "heapster")
+    )
+    expect(@containergroup.tags).to contain_exactly(
+      tag_in_category_with_description(@name_category, "heapster")
+    )
 
     # Check the relation to container node
     expect(@containergroup.container_node).not_to be_nil
@@ -160,6 +174,9 @@ def assert_specific_container_group
     expect(@containergroup.container_replicator).to eq(
       ContainerReplicator.find_by(:name => "monitoring-heapster-controller")
     )
+    expect(@containergroup.container_replicator.labels).to contain_exactly(
+      label_with_name_value("name", "heapster")
+    )
     expect(@containergroup.ext_management_system).to eq(@ems)
 
     # Check pod condition name is "Ready" with status "True"
@@ -189,7 +206,9 @@ def assert_specific_container_node
       :status => "True"
     )
 
-    expect(@containernode.labels.count).to eq(1)
+    expect(@containernode.labels).to contain_exactly(
+      label_with_name_value("kubernetes.io/hostname", "10.35.0.169")
+    )
 
     expect(@containernode.computer_system.operating_system).to have_attributes(
       :distribution   => "Fedora 20 (Heisenbug)",
@@ -226,7 +245,10 @@ def assert_specific_container_service
       :session_affinity => "None",
       :portal_ip        => "10.0.0.1",
     )
-    expect(@containersrv.labels.count).to eq(2)
+    expect(@containersrv.labels).to contain_exactly(
+      label_with_name_value("provider", "kubernetes"),
+      label_with_name_value("component", "apiserver")
+    )
     expect(@containersrv.selector_parts.count).to eq(0)
 
     @confs = @containersrv.container_service_port_configs
@@ -262,7 +284,12 @@ def assert_specific_container_replicator
       :replicas         => 1,
       :current_replicas => 1
     )
-    expect(@replicator.labels.count).to eq(1)
+    expect(@replicator.labels).to contain_exactly(
+      label_with_name_value("name", "influxGrafana")
+    )
+    expect(@replicator.tags).to contain_exactly(
+      tag_in_category_with_description(@name_category, "influxGrafana")
+    )
     expect(@replicator.selector_parts.count).to eq(1)
 
     @group = ContainerGroup.where(:name => "monitoring-influx-grafana-controller-22icy").first
@@ -345,4 +372,15 @@ def assert_specific_container_component_status
       :status    => "True"
     )
   end
+
+  def label_with_name_value(name, value)
+    an_object_having_attributes(
+      :section => 'labels', :source => 'kubernetes',
+      :name => name, :value => value
+    )
+  end
+
+  def tag_in_category_with_description(category, description)
+    satisfy { |tag| tag.category == category && tag.classification.description == description }
+  end
 end
diff --git a/spec/models/manageiq/providers/openshift/container_manager/refresh_parser_spec.rb b/spec/models/manageiq/providers/openshift/container_manager/refresh_parser_spec.rb
index f23c6411afb..c4bafe99d98 100644
--- a/spec/models/manageiq/providers/openshift/container_manager/refresh_parser_spec.rb
+++ b/spec/models/manageiq/providers/openshift/container_manager/refresh_parser_spec.rb
@@ -48,7 +48,8 @@
                                  :output_name                 => 'spec_output_to_name',
 
                                  :completion_deadline_seconds => '11',
-                                 :labels_and_tags             => []
+                                 :labels                      => [],
+                                 :tags                        => []
                                 )
     end
   end