diff --git a/lib/datadog/appsec/contrib/rack/gateway/watcher.rb b/lib/datadog/appsec/contrib/rack/gateway/watcher.rb
index 87c2e7da2ff..603f99793ef 100644
--- a/lib/datadog/appsec/contrib/rack/gateway/watcher.rb
+++ b/lib/datadog/appsec/contrib/rack/gateway/watcher.rb
@@ -27,7 +27,6 @@ def self.watch
                   # TODO: factor out
                   if defined?(Datadog::Tracing) && Datadog::Tracing.respond_to?(:active_span)
                     active_trace = Datadog::Tracing.active_trace
-                    root_span = active_trace.send(:root_span) if active_trace
                     active_span = Datadog::Tracing.active_span
 
                     Datadog.logger.debug { "active span: #{active_span.span_id}" } if active_span
@@ -45,7 +44,6 @@ def self.watch
                       event = {
                         waf_result: result,
                         trace: active_trace,
-                        root_span: root_span,
                         span: active_span,
                         request: request,
                         action: action
@@ -77,7 +75,6 @@ def self.watch
                   # TODO: factor out
                   if defined?(Datadog::Tracing) && Datadog::Tracing.respond_to?(:active_span)
                     active_trace = Datadog::Tracing.active_trace
-                    root_span = active_trace.send(:root_span) if active_trace
                     active_span = Datadog::Tracing.active_span
 
                     Datadog.logger.debug { "active span: #{active_span.span_id}" } if active_span
@@ -95,7 +92,6 @@ def self.watch
                       event = {
                         waf_result: result,
                         trace: active_trace,
-                        root_span: root_span,
                         span: active_span,
                         response: response,
                         action: action
diff --git a/lib/datadog/appsec/event.rb b/lib/datadog/appsec/event.rb
index 02b5895757f..d12bd63de3c 100644
--- a/lib/datadog/appsec/event.rb
+++ b/lib/datadog/appsec/event.rb
@@ -43,26 +43,20 @@ def self.record(*events)
       end
 
       # rubocop:disable Metrics/AbcSize
-      # rubocop:disable Metrics/MethodLength
       def self.record_via_span(*events)
-        events.group_by { |e| e[:root_span] }.each do |root_span, event_group|
-          unless root_span
-            Datadog.logger.debug { "{ error: 'no root span: cannot record', event_group: #{event_group.inspect}}" }
+        events.group_by { |e| e[:trace] }.each do |trace, event_group|
+          unless trace
+            Datadog.logger.debug { "{ error: 'no trace: cannot record', event_group: #{event_group.inspect}}" }
             next
           end
 
-          # TODO: this is a hack but there is no API to do that
-          root_span_tags = root_span.send(:meta).keys
+          trace.keep!
 
           # prepare and gather tags to apply
-          root_tags = event_group.each_with_object({}) do |event, tags|
+          trace_tags = event_group.each_with_object({}) do |event, tags|
             span = event[:span]
-            trace = event[:trace]
 
-            if span
-              span.set_tag('appsec.event', 'true')
-              trace.keep!
-            end
+            span.set_tag('appsec.event', 'true') if span
 
             request = event[:request]
             response = event[:response]
@@ -98,16 +92,15 @@ def self.record_via_span(*events)
           # apply tags to root span
 
           # complex types are unsupported, we need to serialize to a string
-          triggers = root_tags.delete('_dd.appsec.triggers')
-          root_span.set_tag('_dd.appsec.json', JSON.dump({ triggers: triggers }))
+          triggers = trace_tags.delete('_dd.appsec.triggers')
+          trace.set_tag('_dd.appsec.json', JSON.dump({ triggers: triggers }))
 
-          root_tags.each do |key, value|
-            root_span.set_tag(key, value.is_a?(String) ? value.encode('UTf-8') : value) unless root_span_tags.include?(key)
+          trace_tags.each do |key, value|
+            trace.set_tag(key, value)
           end
         end
       end
       # rubocop:enable Metrics/AbcSize
-      # rubocop:enable Metrics/MethodLength
     end
   end
 end
diff --git a/lib/datadog/core/utils/safe_dup.rb b/lib/datadog/core/utils/safe_dup.rb
new file mode 100644
index 00000000000..13ab86a77fb
--- /dev/null
+++ b/lib/datadog/core/utils/safe_dup.rb
@@ -0,0 +1,27 @@
+# typed: false
+
+module Datadog
+  module Core
+    module Utils
+      # Helper methods for safer dup
+      module SafeDup
+        if RUBY_VERSION < '2.2' # nil.dup only fails in Ruby 2.1
+          # Ensures #initialize can call nil.dup safely
+          module RefineNil
+            refine NilClass do
+              def dup
+                self
+              end
+            end
+          end
+
+          using RefineNil
+        end
+
+        def self.frozen_or_dup(v)
+          v.frozen? ? v : v.dup
+        end
+      end
+    end
+  end
+end
diff --git a/lib/datadog/tracing/metadata.rb b/lib/datadog/tracing/metadata.rb
index bf57c110ae0..3ed206de476 100644
--- a/lib/datadog/tracing/metadata.rb
+++ b/lib/datadog/tracing/metadata.rb
@@ -2,6 +2,7 @@
 
 require 'datadog/tracing/metadata/analytics'
 require 'datadog/tracing/metadata/tagging'
+require 'datadog/tracing/metadata/errors'
 
 module Datadog
   module Tracing
@@ -9,6 +10,7 @@ module Tracing
     module Metadata
       def self.included(base)
         base.include(Metadata::Tagging)
+        base.include(Metadata::Errors)
 
         # Additional extensions
         base.prepend(Metadata::Analytics)
diff --git a/lib/datadog/tracing/metadata/errors.rb b/lib/datadog/tracing/metadata/errors.rb
new file mode 100644
index 00000000000..a24e1ca634d
--- /dev/null
+++ b/lib/datadog/tracing/metadata/errors.rb
@@ -0,0 +1,24 @@
+# typed: false
+
+require 'datadog/core/error'
+
+require 'datadog/tracing/metadata/ext'
+
+module Datadog
+  module Tracing
+    module Metadata
+      # Adds error tagging behavior
+      # @public_api
+      module Errors
+        # Mark the span with the given error.
+        def set_error(e)
+          e = Core::Error.build_from(e)
+
+          set_tag(Ext::Errors::TAG_TYPE, e.type) unless e.type.empty?
+          set_tag(Ext::Errors::TAG_MSG, e.message) unless e.message.empty?
+          set_tag(Ext::Errors::TAG_STACK, e.backtrace) unless e.backtrace.empty?
+        end
+      end
+    end
+  end
+end
diff --git a/lib/datadog/tracing/metadata/tagging.rb b/lib/datadog/tracing/metadata/tagging.rb
index 5d21308cfee..12a66555cef 100644
--- a/lib/datadog/tracing/metadata/tagging.rb
+++ b/lib/datadog/tracing/metadata/tagging.rb
@@ -1,6 +1,5 @@
 # typed: false
 
-require 'datadog/core/error'
 require 'datadog/core/environment/ext'
 
 require 'datadog/tracing/metadata/ext'
@@ -96,15 +95,6 @@ def clear_metric(key)
           metrics.delete(key)
         end
 
-        # Mark the span with the given error.
-        def set_error(e)
-          e = Core::Error.build_from(e)
-
-          set_tag(Ext::Errors::TAG_TYPE, e.type) unless e.type.empty?
-          set_tag(Ext::Errors::TAG_MSG, e.message) unless e.message.empty?
-          set_tag(Ext::Errors::TAG_STACK, e.backtrace) unless e.backtrace.empty?
-        end
-
         protected
 
         def meta
diff --git a/lib/datadog/tracing/trace_operation.rb b/lib/datadog/tracing/trace_operation.rb
index b739002b59d..8bdcd39ffd8 100644
--- a/lib/datadog/tracing/trace_operation.rb
+++ b/lib/datadog/tracing/trace_operation.rb
@@ -9,6 +9,7 @@
 require 'datadog/tracing/span_operation'
 require 'datadog/tracing/trace_segment'
 require 'datadog/tracing/trace_digest'
+require 'datadog/tracing/metadata/tagging'
 
 module Datadog
   module Tracing
@@ -24,6 +25,8 @@ module Tracing
     # rubocop:disable Metrics/ClassLength
     # @public_api
     class TraceOperation
+      include Metadata::Tagging
+
       DEFAULT_MAX_LENGTH = 100_000
 
       attr_accessor \
@@ -63,7 +66,9 @@ def initialize(
         sample_rate: nil,
         sampled: nil,
         sampling_priority: nil,
-        service: nil
+        service: nil,
+        tags: nil,
+        metrics: nil
       )
         # Attributes
         @events = events || Events.new
@@ -84,6 +89,10 @@ def initialize(
         @sampling_priority = sampling_priority
         @service = service
 
+        # Generic tags
+        set_tags(tags) if tags
+        set_tags(metrics) if metrics
+
         # State
         @root_span = nil
         @active_span = nil
@@ -239,7 +248,7 @@ def to_digest
           trace_resource: @resource,
           trace_runtime_id: Core::Environment::Identity.id,
           trace_sampling_priority: @sampling_priority,
-          trace_service: @service
+          trace_service: @service,
         ).freeze
       end
 
@@ -261,7 +270,9 @@ def fork_clone
           sample_rate: @sample_rate,
           sampled: @sampled,
           sampling_priority: @sampling_priority,
-          service: (@service && @service.dup)
+          service: (@service && @service.dup),
+          tags: meta.dup,
+          metrics: metrics.dup
         )
       end
 
@@ -404,6 +415,8 @@ def build_trace(spans, partial = false)
           name: @name,
           resource: @resource,
           service: @service,
+          tags: meta,
+          metrics: metrics,
           root_span_id: !partial ? @root_span && @root_span.id : nil
         )
       end
diff --git a/lib/datadog/tracing/trace_segment.rb b/lib/datadog/tracing/trace_segment.rb
index 46f2508fc11..a07fbdf1546 100644
--- a/lib/datadog/tracing/trace_segment.rb
+++ b/lib/datadog/tracing/trace_segment.rb
@@ -1,9 +1,11 @@
 # typed: true
 
 require 'datadog/core/runtime/ext'
+require 'datadog/core/utils/safe_dup'
 
 require 'datadog/tracing/sampling/ext'
 require 'datadog/tracing/metadata/ext'
+require 'datadog/tracing/metadata/tagging'
 
 module Datadog
   module Tracing
@@ -18,21 +20,22 @@ class TraceSegment
       attr_reader \
         :id,
         :spans,
-        :tags
-
-      if RUBY_VERSION < '2.2' # nil.dup only fails in Ruby 2.1
-        # Ensures #initialize can call nil.dup safely
-        module RefineNil
-          refine NilClass do
-            def dup
-              self
-            end
-          end
-        end
-
-        using RefineNil
-      end
-
+        :agent_sample_rate,
+        :hostname,
+        :lang,
+        :name,
+        :origin,
+        :process_id,
+        :rate_limiter_rate,
+        :resource,
+        :rule_sample_rate,
+        :runtime_id,
+        :sample_rate,
+        :sampling_priority,
+        :service
+
+      # rubocop:disable Metrics/CyclomaticComplexity
+      # rubocop:disable Metrics/PerceivedComplexity
       def initialize(
         spans,
         agent_sample_rate: nil,
@@ -49,28 +52,36 @@ def initialize(
         runtime_id: nil,
         sample_rate: nil,
         sampling_priority: nil,
-        service: nil
+        service: nil,
+        tags: nil,
+        metrics: nil
       )
         @id = id
         @root_span_id = root_span_id
         @spans = spans || []
-        @tags = {}
-
-        # Set well-known tags
-        self.agent_sample_rate = agent_sample_rate
-        self.hostname = hostname
-        self.lang = lang
-        self.name = (name.frozen? ? name : name.dup)
-        self.origin = (origin.frozen? ? origin : origin.dup)
-        self.process_id = process_id
-        self.rate_limiter_rate = rate_limiter_rate
-        self.resource = (resource.frozen? ? resource : resource.dup)
-        self.rule_sample_rate = rule_sample_rate
-        self.runtime_id = runtime_id
-        self.sample_rate = sample_rate
-        self.sampling_priority = sampling_priority
-        self.service = (service.frozen? ? service : service.dup)
-      end
+
+        # Does not make an effort to move metrics out of tags
+        # The caller is expected to have done that
+        @meta = tags || {}
+        @metrics = metrics || {}
+
+        # Set well-known tags, defaulting to getting the values from tags
+        @agent_sample_rate = agent_sample_rate || agent_sample_rate_tag
+        @hostname = hostname || hostname_tag
+        @lang = lang || lang_tag
+        @name = Core::Utils::SafeDup.frozen_or_dup(name || name_tag)
+        @origin = Core::Utils::SafeDup.frozen_or_dup(origin || origin_tag)
+        @process_id = process_id || process_id_tag
+        @rate_limiter_rate = rate_limiter_rate || rate_limiter_rate_tag
+        @resource = Core::Utils::SafeDup.frozen_or_dup(resource || resource_tag)
+        @rule_sample_rate = rule_sample_rate_tag || rule_sample_rate
+        @runtime_id = runtime_id || runtime_id_tag
+        @sample_rate = sample_rate || sample_rate_tag
+        @sampling_priority = sampling_priority || sampling_priority_tag
+        @service = Core::Utils::SafeDup.frozen_or_dup(service || service_tag)
+      end
+      # rubocop:enable Metrics/PerceivedComplexity
+      # rubocop:enable Metrics/CyclomaticComplexity
 
       def any?
         @spans.any?
@@ -92,202 +103,104 @@ def size
         @spans.size
       end
 
-      def agent_sample_rate
-        tags[Metadata::Ext::Sampling::TAG_AGENT_RATE]
-      end
-
-      def agent_sample_rate=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::Sampling::TAG_AGENT_RATE)
-          return
-        end
-
-        tags[Metadata::Ext::Sampling::TAG_AGENT_RATE] = value
-      end
-
-      def hostname
-        tags[Metadata::Ext::NET::TAG_HOSTNAME]
-      end
-
-      def hostname=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::NET::TAG_HOSTNAME)
-          return
-        end
-
-        tags[Metadata::Ext::NET::TAG_HOSTNAME] = value
-      end
-
-      def lang
-        tags[Core::Runtime::Ext::TAG_LANG]
-      end
-
-      def lang=(value)
-        if value.nil?
-          tags.delete(Core::Runtime::Ext::TAG_LANG)
-          return
-        end
-
-        tags[Core::Runtime::Ext::TAG_LANG] = value
-      end
-
-      def name
-        tags[TAG_NAME]
-      end
-
-      def name=(value)
-        if value.nil?
-          tags.delete(TAG_NAME)
-          return
-        end
-
-        tags[TAG_NAME] = value
-      end
-
-      def origin
-        tags[Metadata::Ext::Distributed::TAG_ORIGIN]
-      end
-
-      def origin=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::Distributed::TAG_ORIGIN)
-          return
-        end
-
-        tags[Metadata::Ext::Distributed::TAG_ORIGIN] = value
-      end
-
-      def process_id
-        tags[Core::Runtime::Ext::TAG_PID]
+      # If an active trace is present, forces it to be retained by the Datadog backend.
+      #
+      # Any sampling logic will not be able to change this decision.
+      #
+      # @return [void]
+      def keep!
+        self.sampling_priority = Sampling::Ext::Priority::USER_KEEP
       end
 
-      def process_id=(value)
-        if value.nil?
-          tags.delete(Core::Runtime::Ext::TAG_PID)
-          return
-        end
-
-        tags[Core::Runtime::Ext::TAG_PID] = value
+      # If an active trace is present, forces it to be dropped and not stored by the Datadog backend.
+      #
+      # Any sampling logic will not be able to change this decision.
+      #
+      # @return [void]
+      def reject!
+        self.sampling_priority = Sampling::Ext::Priority::USER_REJECT
       end
 
-      def rate_limiter_rate
-        tags[Metadata::Ext::Sampling::TAG_RATE_LIMITER_RATE]
+      def sampled?
+        sampling_priority == Sampling::Ext::Priority::AUTO_KEEP \
+          || sampling_priority == Sampling::Ext::Priority::USER_KEEP
       end
 
-      def rate_limiter_rate=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::Sampling::TAG_RATE_LIMITER_RATE)
-          return
-        end
+      protected
 
-        tags[Metadata::Ext::Sampling::TAG_RATE_LIMITER_RATE] = value
-      end
+      attr_reader \
+        :root_span_id,
+        :meta,
+        :metrics
 
-      def resource
-        tags[TAG_RESOURCE]
-      end
+      private
 
-      def resource=(value)
-        if value.nil?
-          tags.delete(TAG_RESOURCE)
-          return
-        end
+      attr_writer \
+        :agent_sample_rate,
+        :hostname,
+        :lang,
+        :name,
+        :origin,
+        :process_id,
+        :rate_limiter_rate,
+        :resource,
+        :rule_sample_rate,
+        :runtime_id,
+        :sample_rate,
+        :sampling_priority,
+        :service
 
-        tags[TAG_RESOURCE] = value
+      def agent_sample_rate_tag
+        metrics[Metadata::Ext::Sampling::TAG_AGENT_RATE]
       end
 
-      def rule_sample_rate
-        tags[Metadata::Ext::Sampling::TAG_RULE_SAMPLE_RATE]
+      def hostname_tag
+        meta[Metadata::Ext::NET::TAG_HOSTNAME]
       end
 
-      def rule_sample_rate=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::Sampling::TAG_RULE_SAMPLE_RATE)
-          return
-        end
-
-        tags[Metadata::Ext::Sampling::TAG_RULE_SAMPLE_RATE] = value
+      def lang_tag
+        meta[Core::Runtime::Ext::TAG_LANG]
       end
 
-      def runtime_id
-        tags[Core::Runtime::Ext::TAG_ID]
+      def name_tag
+        meta[TAG_NAME]
       end
 
-      def runtime_id=(value)
-        if value.nil?
-          tags.delete(Core::Runtime::Ext::TAG_ID)
-          return
-        end
-
-        tags[Core::Runtime::Ext::TAG_ID] = value
+      def origin_tag
+        meta[Metadata::Ext::Distributed::TAG_ORIGIN]
       end
 
-      def sample_rate
-        tags[Metadata::Ext::Sampling::TAG_SAMPLE_RATE]
+      def process_id_tag
+        meta[Core::Runtime::Ext::TAG_PID]
       end
 
-      def sample_rate=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::Sampling::TAG_SAMPLE_RATE)
-          return
-        end
-
-        tags[Metadata::Ext::Sampling::TAG_SAMPLE_RATE] = value
+      def rate_limiter_rate_tag
+        metrics[Metadata::Ext::Sampling::TAG_RATE_LIMITER_RATE]
       end
 
-      def sampling_priority
-        tags[Metadata::Ext::Distributed::TAG_SAMPLING_PRIORITY]
+      def resource_tag
+        meta[TAG_RESOURCE]
       end
 
-      def sampling_priority=(value)
-        if value.nil?
-          tags.delete(Metadata::Ext::Distributed::TAG_SAMPLING_PRIORITY)
-          return
-        end
-
-        tags[Metadata::Ext::Distributed::TAG_SAMPLING_PRIORITY] = value
+      def rule_sample_rate_tag
+        metrics[Metadata::Ext::Sampling::TAG_RULE_SAMPLE_RATE]
       end
 
-      def service
-        tags[TAG_SERVICE]
+      def runtime_id_tag
+        meta[Core::Runtime::Ext::TAG_ID]
       end
 
-      def service=(value)
-        if value.nil?
-          tags.delete(TAG_SERVICE)
-          return
-        end
-
-        tags[TAG_SERVICE] = value
+      def sample_rate_tag
+        metrics[Metadata::Ext::Sampling::TAG_SAMPLE_RATE]
       end
 
-      # If an active trace is present, forces it to be retained by the Datadog backend.
-      #
-      # Any sampling logic will not be able to change this decision.
-      #
-      # @return [void]
-      def keep!
-        self.sampling_priority = Sampling::Ext::Priority::USER_KEEP
+      def sampling_priority_tag
+        meta[Metadata::Ext::Distributed::TAG_SAMPLING_PRIORITY]
       end
 
-      # If an active trace is present, forces it to be dropped and not stored by the Datadog backend.
-      #
-      # Any sampling logic will not be able to change this decision.
-      #
-      # @return [void]
-      def reject!
-        self.sampling_priority = Sampling::Ext::Priority::USER_REJECT
+      def service_tag
+        meta[TAG_SERVICE]
       end
-
-      def sampled?
-        sampling_priority == Sampling::Ext::Priority::AUTO_KEEP \
-          || sampling_priority == Sampling::Ext::Priority::USER_KEEP
-      end
-
-      protected
-
-      attr_reader \
-        :root_span_id
     end
     # rubocop:enable Metrics/ClassLength
   end
diff --git a/lib/ddtrace/transport/trace_formatter.rb b/lib/ddtrace/transport/trace_formatter.rb
index 567638817ff..c413075a3d0 100644
--- a/lib/ddtrace/transport/trace_formatter.rb
+++ b/lib/ddtrace/transport/trace_formatter.rb
@@ -32,6 +32,11 @@ def format!
         # trace metadata, we must put our trace
         # metadata on the root span. This "root span"
         # is needed by the agent/API to ingest the trace.
+
+        # Apply generic trace tags. Any more specific value will be overridden
+        # by the subsequent calls below.
+        set_trace_tags!
+
         set_resource!
 
         tag_agent_sample_rate!
@@ -54,11 +59,21 @@ def set_resource!
         # If the trace resource is undefined, or the root span wasn't
         # specified, don't set this. We don't want to overwrite the
         # resource of a span that is in the middle of the trace.
-        return if trace.resource.nil? || !@found_root_span
+        return if trace.resource.nil? || partial?
 
         root_span.resource = trace.resource
       end
 
+      def set_trace_tags!
+        # If the root span wasn't specified, don't set this. We don't want to
+        # misset or overwrite the tags of a span that is in the middle of the
+        # trace.
+        return if partial?
+
+        root_span.set_tags(trace.send(:meta))
+        root_span.set_tags(trace.send(:metrics))
+      end
+
       def tag_agent_sample_rate!
         return unless trace.agent_sample_rate
 
@@ -151,6 +166,10 @@ def tag_sampling_priority!
 
       private
 
+      def partial?
+        !@found_root_span
+      end
+
       def find_root_span(trace)
         # TODO: Should we memoize this?
         #       Would be safe, but `spans` is mutable, so if
@@ -159,6 +178,8 @@ def find_root_span(trace)
         root_span_id = trace.send(:root_span_id)
         root_span = trace.spans.find { |s| s.id == root_span_id } if root_span_id
         @found_root_span = !root_span.nil?
+
+        # when root span is not found, fall back to last span (partial flush)
         root_span || trace.spans.last
       end
     end
diff --git a/spec/datadog/tracing/metadata/errors_spec.rb b/spec/datadog/tracing/metadata/errors_spec.rb
new file mode 100644
index 00000000000..e312644a7eb
--- /dev/null
+++ b/spec/datadog/tracing/metadata/errors_spec.rb
@@ -0,0 +1,35 @@
+# typed: ignore
+
+require 'spec_helper'
+
+require 'datadog/tracing/metadata/tagging'
+require 'datadog/tracing/metadata/errors'
+
+RSpec.describe Datadog::Tracing::Metadata::Errors do
+  subject(:test_object) { test_class.new }
+  let(:test_class) do
+    Class.new do
+      include Datadog::Tracing::Metadata::Tagging
+      include Datadog::Tracing::Metadata::Errors
+    end
+  end
+
+  describe '#set_error' do
+    subject(:set_error) { test_object.set_error(error) }
+
+    let(:error) { RuntimeError.new('oops') }
+    let(:backtrace) { %w[method1 method2 method3] }
+
+    before { error.set_backtrace(backtrace) }
+
+    it do
+      set_error
+
+      expect(test_object).to have_error_message('oops')
+      expect(test_object).to have_error_type('RuntimeError')
+      backtrace.each do |method|
+        expect(test_object).to have_error_stack(include(method))
+      end
+    end
+  end
+end
diff --git a/spec/datadog/tracing/metadata/tagging_spec.rb b/spec/datadog/tracing/metadata/tagging_spec.rb
index bf78b6ed323..30990ab735e 100644
--- a/spec/datadog/tracing/metadata/tagging_spec.rb
+++ b/spec/datadog/tracing/metadata/tagging_spec.rb
@@ -352,23 +352,4 @@
       expect(test_object.send(:metrics)).to_not have_key(key)
     end
   end
-
-  describe '#set_error' do
-    subject(:set_error) { test_object.set_error(error) }
-
-    let(:error) { RuntimeError.new('oops') }
-    let(:backtrace) { %w[method1 method2 method3] }
-
-    before { error.set_backtrace(backtrace) }
-
-    it do
-      set_error
-
-      expect(test_object).to have_error_message('oops')
-      expect(test_object).to have_error_type('RuntimeError')
-      backtrace.each do |method|
-        expect(test_object).to have_error_stack(include(method))
-      end
-    end
-  end
 end
diff --git a/spec/datadog/tracing/metadata_spec.rb b/spec/datadog/tracing/metadata_spec.rb
index 33d59a96690..4a30cce1beb 100644
--- a/spec/datadog/tracing/metadata_spec.rb
+++ b/spec/datadog/tracing/metadata_spec.rb
@@ -3,8 +3,6 @@
 require 'spec_helper'
 
 require 'datadog/tracing/metadata'
-require 'datadog/tracing/metadata/analytics'
-require 'datadog/tracing/metadata/tagging'
 
 RSpec.describe Datadog::Tracing::Metadata do
   context 'when included' do
@@ -16,7 +14,8 @@
       it 'has all of the tagging behavior in correct order' do
         expect(ancestors.first(5)).to include(
           described_class::Analytics,
-          described_class::Tagging
+          described_class::Tagging,
+          described_class::Errors
         )
       end
     end
diff --git a/spec/datadog/tracing/trace_operation_spec.rb b/spec/datadog/tracing/trace_operation_spec.rb
index ca94aa5df55..dda4ee98d64 100644
--- a/spec/datadog/tracing/trace_operation_spec.rb
+++ b/spec/datadog/tracing/trace_operation_spec.rb
@@ -30,7 +30,9 @@
         sample_rate: sample_rate,
         sampled: sampled,
         sampling_priority: sampling_priority,
-        service: service
+        service: service,
+        tags: tags,
+        metrics: metrics
       }
     end
 
@@ -46,6 +48,8 @@
     let(:sampled) { true }
     let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::USER_KEEP }
     let(:service) { 'billing-api' }
+    let(:tags) { { 'foo' => 'bar' } }
+    let(:metrics) { { 'baz' => 42.0 } }
   end
 
   shared_examples 'a span with default events' do
@@ -73,6 +77,14 @@
           service: nil
         )
       end
+
+      it do
+        expect(trace_op.send(:meta)).to eq({})
+      end
+
+      it do
+        expect(trace_op.send(:metrics)).to eq({})
+      end
     end
 
     context 'given' do
@@ -173,6 +185,20 @@
 
         it { expect(trace_op.service).to eq(service) }
       end
+
+      context ':tags' do
+        subject(:options) { { tags: tags } }
+        let(:tags) { { 'foo' => 'bar' } }
+
+        it { expect(trace_op.send(:meta)).to eq({ 'foo' => 'bar' }) }
+      end
+
+      context ':metrics' do
+        subject(:options) { { metrics: metrics } }
+        let(:metrics) { { 'baz' => 42.0 } }
+
+        it { expect(trace_op.send(:metrics)).to eq({ 'baz' => 42.0 }) }
+      end
     end
   end
 
@@ -775,6 +801,107 @@
     end
   end
 
+  describe '#get_tag' do
+    before do
+      trace_op.set_tag('foo', 'bar')
+    end
+
+    it 'gets tag set on trace' do
+      expect(trace_op.get_tag('foo')).to eq('bar')
+    end
+
+    it 'gets unset tag as nil' do
+      expect(trace_op.get_tag('unset')).to be_nil
+    end
+  end
+
+  describe '#set_metric' do
+    it 'sets metrics' do
+      trace_op.set_metric('foo', 42)
+      trace_op.measure('top') {}
+
+      trace = trace_op.flush!
+
+      expect(trace.send(:metrics)['foo']).to eq(42)
+    end
+  end
+
+  describe '#set_tag' do
+    it 'sets tag on trace before a measurement' do
+      trace_op.set_tag('foo', 'bar')
+      trace_op.measure('top') {}
+
+      trace = trace_op.flush!
+
+      expect(trace.send(:meta)['foo']).to eq('bar')
+    end
+
+    it 'sets tag on trace after a measurement' do
+      trace_op.measure('top') {}
+      trace_op.set_tag('foo', 'bar')
+
+      trace = trace_op.flush!
+
+      expect(trace.send(:meta)['foo']).to eq('bar')
+    end
+
+    it 'sets tag on trace from a measurement' do
+      trace_op.measure('top') do
+        trace_op.set_tag('foo', 'bar')
+      end
+
+      trace = trace_op.flush!
+
+      expect(trace.send(:meta)['foo']).to eq('bar')
+    end
+
+    it 'sets tag on trace from a nested measurement' do
+      trace_op.measure('grandparent') do
+        trace_op.measure('parent') do
+          trace_op.set_tag('foo', 'bar')
+        end
+      end
+
+      trace = trace_op.flush!
+
+      expect(trace.spans).to have(2).items
+      expect(trace.spans.map(&:name)).to include('parent')
+      expect(trace.send(:meta)['foo']).to eq('bar')
+    end
+
+    it 'sets metrics' do
+      trace_op.set_tag('foo', 42)
+      trace_op.measure('top') {}
+
+      trace = trace_op.flush!
+
+      expect(trace.send(:metrics)['foo']).to eq(42)
+    end
+
+    context 'with partial flushing' do
+      subject(:flush!) { trace_op.flush! }
+      let(:trace) { flush! }
+
+      it 'sets tag on trace from a nested measurement' do
+        trace_op.measure('grandparent') do
+          trace_op.measure('parent') do
+            trace_op.set_tag('foo', 'bar')
+          end
+          flush!
+        end
+
+        expect(trace.spans).to have(1).items
+        expect(trace.spans.map(&:name)).to include('parent')
+        expect(trace.send(:meta)['foo']).to eq('bar')
+
+        final_flush = trace_op.flush!
+        expect(final_flush.spans).to have(1).items
+        expect(final_flush.spans.map(&:name)).to include('grandparent')
+        expect(final_flush.send(:meta)['foo']).to eq('bar')
+      end
+    end
+  end
+
   describe '#build_span' do
     subject(:build_span) { trace_op.build_span(span_name, **span_options) }
     let(:span_name) { 'web.request' }
@@ -1669,6 +1796,14 @@ def span
             )
           end
 
+          it 'maintains the same tags' do
+            expect(new_trace_op.send(:meta)).to eq({ 'foo' => 'bar' })
+          end
+
+          it 'maintains the same metrics' do
+            expect(new_trace_op.send(:metrics)).to eq({ 'baz' => 42.0 })
+          end
+
           it 'maintains the same events' do
             old_events = trace_op.send(:events)
             new_events = new_trace_op.send(:events)
@@ -1731,6 +1866,14 @@ def span
           )
         end
 
+        it 'maintains the same tags' do
+          expect(new_trace_op.send(:meta)).to eq({ 'foo' => 'bar' })
+        end
+
+        it 'maintains the same metrics' do
+          expect(new_trace_op.send(:metrics)).to eq({ 'baz' => 42.0 })
+        end
+
         context 'and :parent_span_id has been defined' do
           let(:options) { { parent_span_id: parent_span_id } }
           let(:parent_span_id) { Datadog::Core::Utils.next_id }
@@ -1771,6 +1914,14 @@ def span
               service: be_a_copy_of(service)
             )
           end
+
+          it 'maintains the same tags' do
+            expect(new_trace_op.send(:meta)).to eq({ 'foo' => 'bar' })
+          end
+
+          it 'maintains the same metrics' do
+            expect(new_trace_op.send(:metrics)).to eq({ 'baz' => 42.0 })
+          end
         end
 
         context 'that has started' do
@@ -1802,6 +1953,14 @@ def span
               service: be_a_copy_of(service)
             )
           end
+
+          it 'maintains the same tags' do
+            expect(new_trace_op.send(:meta)).to eq({ 'foo' => 'bar' })
+          end
+
+          it 'maintains the same metrics' do
+            expect(new_trace_op.send(:metrics)).to eq({ 'baz' => 42.0 })
+          end
         end
 
         context 'that has finished' do
@@ -1833,6 +1992,14 @@ def span
               service: be_a_copy_of(service)
             )
           end
+
+          it 'maintains the same tags' do
+            expect(new_trace_op.send(:meta)).to eq({ 'foo' => 'bar' })
+          end
+
+          it 'maintains the same metrics' do
+            expect(new_trace_op.send(:metrics)).to eq({ 'baz' => 42.0 })
+          end
         end
       end
 
@@ -1875,6 +2042,14 @@ def span
           )
         end
 
+        it 'maintains the same tags' do
+          expect(new_trace_op.send(:meta)).to eq({ 'foo' => 'bar' })
+        end
+
+        it 'maintains the same metrics' do
+          expect(new_trace_op.send(:metrics)).to eq({ 'baz' => 42.0 })
+        end
+
         context 'and :parent_span_id has been defined' do
           let(:options) { { parent_span_id: parent_span_id } }
           let(:parent_span_id) { Datadog::Core::Utils.next_id }
diff --git a/spec/datadog/tracing/trace_segment_spec.rb b/spec/datadog/tracing/trace_segment_spec.rb
index 049b974d765..11c90e354c5 100644
--- a/spec/datadog/tracing/trace_segment_spec.rb
+++ b/spec/datadog/tracing/trace_segment_spec.rb
@@ -45,12 +45,19 @@
           sampling_priority: nil,
           service: nil,
           spans: spans,
-          tags: {}
         )
       end
+
+      it do
+        expect(trace_segment.send(:meta)).to eq({})
+      end
+
+      it do
+        expect(trace_segment.send(:metrics)).to eq({})
+      end
     end
 
-    context 'given' do
+    context 'given arguments' do
       context ':agent_sample_rate' do
         let(:options) { { agent_sample_rate: agent_sample_rate } }
         let(:agent_sample_rate) { rand }
@@ -118,7 +125,7 @@
         let(:options) { { runtime_id: runtime_id } }
         let(:runtime_id) { Datadog::Core::Environment::Identity.id }
 
-        it { is_expected.to have_attributes(runtime_id: be(runtime_id)) }
+        it { is_expected.to have_attributes(runtime_id: runtime_id) }
       end
 
       context ':sample_rate' do
@@ -141,6 +148,119 @@
 
         it { is_expected.to have_attributes(service: be_a_copy_of(service)) }
       end
+
+      context ':tags' do
+        let(:options) { { tags: tags } }
+        let(:tags) { { 'foo' => 'bar' } }
+
+        it { expect(trace_segment.send(:meta)).to eq({ 'foo' => 'bar' }) }
+      end
+
+      context ':metrics' do
+        let(:options) { { metrics: metrics } }
+        let(:metrics) { { 'foo' => 42.0 } }
+
+        it { expect(trace_segment.send(:metrics)).to eq({ 'foo' => 42.0 }) }
+      end
+    end
+
+    context 'given tags' do
+      context ':agent_sample_rate' do
+        let(:options) { { metrics: { Datadog::Tracing::Metadata::Ext::Sampling::TAG_AGENT_RATE => agent_sample_rate } } }
+        let(:agent_sample_rate) { rand }
+
+        it { is_expected.to have_attributes(agent_sample_rate: agent_sample_rate) }
+      end
+
+      context ':hostname' do
+        let(:options) { { tags: { Datadog::Tracing::Metadata::Ext::NET::TAG_HOSTNAME => hostname } } }
+        let(:hostname) { 'my.host' }
+
+        it { is_expected.to have_attributes(hostname: be(hostname)) }
+      end
+
+      context ':lang' do
+        let(:options) { { tags: { Datadog::Core::Runtime::Ext::TAG_LANG => lang } } }
+        let(:lang) { 'ruby' }
+
+        it { is_expected.to have_attributes(lang: be(lang)) }
+      end
+
+      context ':name' do
+        let(:options) { { tags: { Datadog::Tracing::TraceSegment::TAG_NAME => name } } }
+        let(:name) { 'job.work' }
+
+        it { is_expected.to have_attributes(name: be_a_copy_of(name)) }
+      end
+
+      context ':origin' do
+        let(:options) { { tags: { Datadog::Tracing::Metadata::Ext::Distributed::TAG_ORIGIN => origin } } }
+        let(:origin) { 'synthetics' }
+
+        it { is_expected.to have_attributes(origin: be_a_copy_of(origin)) }
+      end
+
+      context ':process_id' do
+        let(:options) { { tags: { Datadog::Core::Runtime::Ext::TAG_PID => process_id } } }
+        let(:process_id) { Datadog::Core::Environment::Identity.pid }
+
+        it { is_expected.to have_attributes(process_id: process_id) }
+      end
+
+      context ':rate_limiter_rate' do
+        let(:options) do
+          { metrics: { Datadog::Tracing::Metadata::Ext::Sampling::TAG_RATE_LIMITER_RATE => rate_limiter_rate } }
+        end
+        let(:rate_limiter_rate) { rand }
+
+        it { is_expected.to have_attributes(rate_limiter_rate: rate_limiter_rate) }
+      end
+
+      context ':resource' do
+        let(:options) { { tags: { Datadog::Tracing::TraceSegment::TAG_RESOURCE => resource } } }
+        let(:resource) { 'generate_report' }
+
+        it { is_expected.to have_attributes(resource: be_a_copy_of(resource)) }
+      end
+
+      context ':rule_sample_rate' do
+        let(:options) do
+          { metrics: { Datadog::Tracing::Metadata::Ext::Sampling::TAG_RULE_SAMPLE_RATE => rule_sample_rate } }
+        end
+        let(:rule_sample_rate) { rand }
+
+        it { is_expected.to have_attributes(rule_sample_rate: rule_sample_rate) }
+      end
+
+      context ':runtime_id' do
+        let(:options) { { tags: { Datadog::Core::Runtime::Ext::TAG_ID => runtime_id } } }
+        let(:runtime_id) { Datadog::Core::Environment::Identity.id }
+
+        it { is_expected.to have_attributes(runtime_id: runtime_id) }
+      end
+
+      context ':sample_rate' do
+        let(:options) { { metrics: { Datadog::Tracing::Metadata::Ext::Sampling::TAG_SAMPLE_RATE => sample_rate } } }
+        let(:sample_rate) { rand }
+
+        it { is_expected.to have_attributes(sample_rate: sample_rate) }
+      end
+
+      context ':sampling_priority' do
+        let(:options) do
+          { tags: { Datadog::Tracing::Metadata::Ext::Distributed::TAG_SAMPLING_PRIORITY => sampling_priority } }
+        end
+        let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::USER_KEEP }
+
+        it { is_expected.to have_attributes(sampling_priority: sampling_priority) }
+      end
+
+      context ':service' do
+        let(:options) { { tags: { Datadog::Tracing::TraceSegment::TAG_SERVICE => service } } }
+        let(:service) { 'job-worker' }
+
+        it { is_expected.to have_attributes(service: be_a_copy_of(service)) }
+      end
     end
   end
 
@@ -185,31 +305,33 @@
 
   describe '#sampled?' do
     subject(:sampled?) { trace_segment.sampled? }
+    let(:options) { { sampling_priority: sampling_priority } }
+    let(:sampling_priority) { nil }
 
     context 'when sampling priority is not set' do
       it { is_expected.to be false }
     end
 
     context 'when sampling priority is set to AUTO_KEEP' do
-      before { trace_segment.sampling_priority = Datadog::Tracing::Sampling::Ext::Priority::AUTO_KEEP }
+      let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::AUTO_KEEP }
 
       it { is_expected.to be true }
     end
 
     context 'when sampling priority is set to USER_KEEP' do
-      before { trace_segment.sampling_priority = Datadog::Tracing::Sampling::Ext::Priority::USER_KEEP }
+      let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::USER_KEEP }
 
       it { is_expected.to be true }
     end
 
     context 'when sampling priority is set to AUTO_REJECT' do
-      before { trace_segment.sampling_priority = Datadog::Tracing::Sampling::Ext::Priority::AUTO_REJECT }
+      let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::AUTO_REJECT }
 
       it { is_expected.to be false }
     end
 
     context 'when sampling priority is set to USER_REJECT' do
-      before { trace_segment.sampling_priority = Datadog::Tracing::Sampling::Ext::Priority::USER_REJECT }
+      let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::USER_REJECT }
 
       it { is_expected.to be false }
     end
diff --git a/spec/ddtrace/transport/trace_formatter_spec.rb b/spec/ddtrace/transport/trace_formatter_spec.rb
index 50f4b1cb309..5bf629ec1f0 100644
--- a/spec/ddtrace/transport/trace_formatter_spec.rb
+++ b/spec/ddtrace/transport/trace_formatter_spec.rb
@@ -16,6 +16,10 @@
   let(:trace_options) { {} }
 
   shared_context 'trace metadata' do
+    let(:trace_tags) do
+      nil
+    end
+
     let(:trace_options) do
       {
         resource: resource,
@@ -28,7 +32,8 @@
         rule_sample_rate: rule_sample_rate,
         runtime_id: runtime_id,
         sample_rate: sample_rate,
-        sampling_priority: sampling_priority
+        sampling_priority: sampling_priority,
+        tags: trace_tags
       }
     end
 
@@ -45,6 +50,17 @@
     let(:sampling_priority) { Datadog::Tracing::Sampling::Ext::Priority::USER_KEEP }
   end
 
+  shared_context 'trace metadata with tags' do
+    include_context 'trace metadata'
+
+    let(:trace_tags) do
+      {
+        'foo' => 'bar',
+        'baz' => 42
+      }
+    end
+  end
+
   shared_context 'no root span' do
     let(:trace) { Datadog::Tracing::TraceSegment.new(spans, **trace_options) }
     let(:spans) { Array.new(3) { Datadog::Tracing::Span.new('my.job') } }
@@ -139,6 +155,26 @@
         end
       end
 
+      shared_examples 'root span with generic tags' do
+        context 'metrics' do
+          it { expect(root_span.metrics).to include({ 'baz' => 42 }) }
+        end
+
+        context 'meta' do
+          it { expect(root_span.meta).to include({ 'foo' => 'bar' }) }
+        end
+      end
+
+      shared_examples 'root span without generic tags' do
+        context 'metrics' do
+          it { expect(root_span.metrics).to_not include({ 'baz' => 42 }) }
+        end
+
+        context 'meta' do
+          it { expect(root_span.meta).to_not include({ 'foo' => 'bar' }) }
+        end
+      end
+
       context 'with no root span' do
         include_context 'no root span'
 
@@ -159,6 +195,16 @@
 
           it_behaves_like 'root span with tags'
         end
+
+        context 'when trace has metadata set with generic tags' do
+          include_context 'trace metadata with tags'
+
+          it { is_expected.to be(trace) }
+          it { expect(root_span.resource).to eq('my.job') }
+
+          it_behaves_like 'root span with tags'
+          it_behaves_like 'root span without generic tags'
+        end
       end
 
       context 'with missing root span' do
@@ -181,6 +227,16 @@
 
           it_behaves_like 'root span with tags'
         end
+
+        context 'when trace has metadata set with generic tags' do
+          include_context 'trace metadata with tags'
+
+          it { is_expected.to be(trace) }
+          it { expect(root_span.resource).to eq('my.job') }
+
+          it_behaves_like 'root span with tags'
+          it_behaves_like 'root span without generic tags'
+        end
       end
 
       context 'with a root span' do
@@ -203,6 +259,16 @@
 
           it_behaves_like 'root span with tags'
         end
+
+        context 'when trace has metadata set with generic tags' do
+          include_context 'trace metadata with tags'
+
+          it { is_expected.to be(trace) }
+          it { expect(root_span.resource).to eq(resource) }
+
+          it_behaves_like 'root span with tags'
+          it_behaves_like 'root span with generic tags'
+        end
       end
     end
   end