Sampling flag propagation

download.sh

@@ -11,19 +11,45 @@
 WORKFLOW_SLUG='https://circleci.com/api/v2/workflow/'
 
 echo "Getting Pipeline ID"
-PIPELINE_ID=$(curl -s \
-                   -H "Circle-Token: ${CIRCLE_TOKEN}" \
-                   "${PROJECT_SLUG}/job/${CIRCLE_BUILD_NUM}" \
-              | jq -rj '.pipeline.id')
+if [[ -n ${WORKFLOW_ID} ]]
+then
+    # Optional: resolve the pipeline from any workflow ID instead of a job number
+    PIPELINE_ID=$(curl -s \
+                       -H "Circle-Token: ${CIRCLE_TOKEN}" \
+                       "${WORKFLOW_SLUG}/${WORKFLOW_ID}" \
+                  | jq -rj '.pipeline_id // empty')
+else
+    PIPELINE_ID=$(curl -s \
+                       -H "Circle-Token: ${CIRCLE_TOKEN}" \
+                       "${PROJECT_SLUG}/job/${CIRCLE_BUILD_NUM}" \
+                  | jq -rj '.pipeline.id // empty')
+fi
 
+if [[ -z ${PIPELINE_ID} ]]
+then
+    echo "ERROR: Could not determine the pipeline ID. Check CIRCLE_TOKEN and CIRCLE_BUILD_NUM (or WORKFLOW_ID)."
+    exit 253
+fi
 
 echo "Received Pipeline ID is: ${PIPELINE_ID}"
 
 echo "Getting Workflows for Pipeline ID"
+# A rerun of a workflow creates a new workflow with the same name inside the
+# same pipeline, so only the most recent run of each workflow name is used.
 WORKFLOW_IDS=$(curl -s \
                     -H "Circle-Token: ${CIRCLE_TOKEN}" \
                     "${PIPELINE_SLUG}/${PIPELINE_ID}/workflow" \
-              | jq -r '.items[] | select(.name!="report_coverage") | .id')
+              | jq -r '.items
+                       | map(select(.name!="report_coverage"))
+                       | group_by(.name)
+                       | map(max_by(.created_at))
+                       | .[].id')
+
+if [[ -z ${WORKFLOW_IDS} ]]
+then
+    echo "ERROR: No workflows found for pipeline ${PIPELINE_ID}"
+    exit 252
+fi
 
 echo "Received Workflow IDs are: ${WORKFLOW_IDS}"
 
@@ -57,13 +83,19 @@
 
 echo "All workflows have successfully finished, downloading artifacts."
 
+rm -rf partial_coverage_results
 mkdir partial_coverage_results
 for workflow_id in ${WORKFLOW_IDS}
 do
+    # If a job was rerun inside the same workflow, only use its latest attempt.
     JOB_NUMBERS=$(curl -s \
                        -H "Circle-Token: ${CIRCLE_TOKEN}" \
                        "${WORKFLOW_SLUG}/${workflow_id}/job" \
-                 | jq -r '.items[] | .job_number' \
+                 | jq -r '.items
+                          | map(select(.job_number != null))
+                          | group_by(.name)
+                          | map(max_by(.job_number))
+                          | .[].job_number' \
                  | tr '\n' ' ')
 
     for job in ${JOB_NUMBERS}
@@ -72,8 +104,14 @@
                                       -H "Circle-Token: ${CIRCLE_TOKEN}" \
                                      "${PROJECT_SLUG}/${job}/artifacts" \
                                  | jq -r \
-                                      '.items[] | select(.path=="coverage/.resultset.json") | .url')
+                                      '(.items // [])[] | select(.path=="coverage/.resultset.json") | .url')
                                       #'.items[] | select(.path=="coverage/coverage.json" or .path=="coverage/.resultset.json") | .url')
+        # Jobs without a coverage artifact (e.g. lint) have nothing to download
+        if [[ -z ${ARTIFACT_DOWNLOAD_LINK} ]]
+        then
+            echo "Skipping job ${job}: no coverage/.resultset.json artifact"
+            continue
+        fi
         curl -s \
              -L \
              -H "Circle-Token: ${CIRCLE_TOKEN}" \

lib/instana/base.rb

@@ -6,6 +6,11 @@
 require "instana/util"
 
 module Instana
+  # Instana trace context header names
+  INSTANA_TRACE_ID_HEADER = 'X-Instana-T'.freeze
+  INSTANA_SPAN_ID_HEADER = 'X-Instana-S'.freeze
+  INSTANA_LEVEL_HEADER = 'X-Instana-L'.freeze
+
   class << self
     attr_accessor :agent
     attr_accessor :tracer

lib/instana/instrumentation/active_job.rb

@@ -1,19 +1,5 @@
 # (c) Copyright IBM Corp. 2021
 # (c) Copyright Instana Inc. 2021
-module OpenTelemetry
-  module Trace
-    module Propagation
-      module TraceContext
-        # A TraceParent is an implementation of the W3C trace context specification
-        # https://www.w3.org/TR/trace-context/
-        # {Trace::SpanContext}
-        class TraceParent
-          REGEXP = /^(?<version>[A-Fa-f0-9]{2})-(?<trace_id>[A-Fa-f0-9]{32})-(?<span_id>[A-Fa-f0-9]{32})-(?<flags>[A-Fa-f0-9]{2})(?<ignored>-.*)?$/
-        end
-      end
-    end
-  end
-end
 
 module Instana
   module Instrumentation
@@ -48,18 +34,62 @@ def self.prepended(target)
               job_id: job.job_id
             }
           }
-          incoming_context = if job.arguments.is_a?(Array) && job.arguments.last.is_a?(Hash) && job.arguments.last.key?(:instana_context)
-                               instana_context = job.arguments.last[:instana_context]
-                               job.arguments.pop
-                               instana_context ? ::Instana::SpanContext.new(trace_id: instana_context[:trace_id], span_id: instana_context[:span_id]) : nil
-                             end
+          incoming_context = ::Instana::Instrumentation::ActiveJob.extract_context(job)
+
           Trace.with_span(OpenTelemetry::Trace.non_recording_span(incoming_context)) do
             ::Instana.tracer.in_span(:activejob, attributes: tags) do
               block.call
             end
           end
         end
       end
+
+      def self.extract_context(job)
+        # 1. From job arguments (Instana payload)
+        if job.arguments.is_a?(Array) && job.arguments.last.is_a?(Hash) && job.arguments.last.key?(:instana_context)
+          instana_context = job.arguments.last[:instana_context]
+          job.arguments.pop
+          return instana_context ? ::Instana::SpanContext.new(
+            trace_id: instana_context[:trace_id],
+            span_id: instana_context[:span_id],
+            level: instana_context[:level] || 1
+          ) : nil
+        end
+
+        # 2. From job.__otel_headers (if opentelemetry-instrumentation-active_job is active)
+        if job.respond_to?(:__otel_headers) && job.__otel_headers.is_a?(Hash)
+          headers = job.__otel_headers
+          tp = headers['traceparent'] || headers[:traceparent]
+          if tp
+            ctx = parse_w3c_traceparent(tp, headers['tracestate'] || headers[:tracestate])
+            return ctx if ctx
+          end
+        end
+
+        nil
+      end
+
+      def self.parse_w3c_traceparent(traceparent, tracestate)
+        m = traceparent.match(/[0-9a-f]{2}-(?<trace>[0-9a-f]{32})-(?<parent>[0-9a-f]{16})-(?<flags>[0-9a-f]{2})/)
+        return nil unless m
+
+        w3c_sampled = (m['flags'].to_i(16) & 0x01) == 1
+        ts_match = tracestate&.match(/in=(?<t>[0-9a-f]+);(?<s>[0-9a-f]+)/)
+
+        if ts_match
+          ::Instana::SpanContext.new(
+            trace_id: ::Instana::Util.header_to_id(ts_match['t']),
+            span_id: ::Instana::Util.header_to_id(ts_match['s']),
+            level: w3c_sampled ? 1 : 0
+          )
+        else
+          ::Instana::SpanContext.new(
+            trace_id: ::Instana::Util.header_to_id(m['trace'][16..]),
+            span_id: ::Instana::Util.header_to_id(m['parent']),
+            level: w3c_sampled ? 1 : 0
+          )
+        end
+      end
     end
   end
 end

lib/instana/instrumentation/aws_sdk_lambda.rb

@@ -9,15 +9,19 @@ class Handler < Seahorse::Client::Handler
         def call(context)
           return @handler.call(context) unless [:invoke_async, :invoke].include?(context.operation_name)
 
-          if context.params[:client_context].nil? && ::Instana.tracer.tracing? && context.operation_name == :invoke
-            span_context = ::Instana.tracer.context
-            payload = {
+          if ::Instana.tracer.tracing? && context.operation_name == :invoke
+            span_context  = ::Instana.tracer.context
+            trace_payload = {
               'X-INSTANA-T' => span_context.trace_id,
               'X-INSTANA-S' => span_context.span_id,
               'X-INSTANA-L' => span_context.level.to_s
             }
-
-            context.params[:client_context] = Base64.strict_encode64(JSON.dump(payload))
+            merged = if context.params[:client_context]
+                       JSON.parse(Base64.strict_decode64(context.params[:client_context])).merge(trace_payload)
+                     else
+                       trace_payload
+                     end
+            context.params[:client_context] = Base64.strict_encode64(JSON.dump(merged))
           end
 
           tags = {

lib/instana/instrumentation/aws_sdk_sns.rb

@@ -14,7 +14,18 @@ def call(context)
           }.reject { |_, v| v.nil? }
 
           if context.operation_name == :publish
-            ::Instana.tracer.in_span(:sns, attributes: {sns: sns_tags}) { @handler.call(context) }
+            ::Instana.tracer.in_span(:sns, attributes: {sns: sns_tags}) do |_span|
+              if ::Instana.tracer.tracing?
+                span_context = ::Instana.tracer.context
+                context.params[:message_attributes] ||= {}
+                context.params[:message_attributes].merge!(
+                  'X_INSTANA_T' => { data_type: 'String', string_value: span_context.trace_id_header },
+                  'X_INSTANA_S' => { data_type: 'String', string_value: span_context.span_id_header },
+                  'X_INSTANA_L' => { data_type: 'String', string_value: span_context.level.to_s }
+                )
+              end
+              @handler.call(context)
+            end
           else
             @handler.call(context)
           end

lib/instana/instrumentation/bunny.rb

@@ -20,9 +20,11 @@ def publish(payload, options = {})
           ::Instana.tracer.in_span(:rabbitmq, attributes: kvs) do |span|
             # Inject trace context into message headers
             options[:headers] ||= {}
-            options[:headers]['X-Instana-T'] = span.context.trace_id
-            options[:headers]['X-Instana-S'] = span.context.span_id
-            options[:headers]['X-Instana-L'] = span.context.level.to_s
+            options[:headers][INSTANA_TRACE_ID_HEADER] = span.context.trace_id
+            options[:headers][INSTANA_SPAN_ID_HEADER] = span.context.span_id
+            options[:headers][INSTANA_LEVEL_HEADER] = span.context.level.to_s
+            options[:headers]['traceparent']  = span.context.trace_parent_header
+            options[:headers]['tracestate']   = span.context.trace_state_header
 
             super(payload, options)
           end
@@ -113,13 +115,42 @@ def extract_context_from_headers(properties)
         return {} unless properties && properties.headers
 
         headers = properties.headers
+
+        if headers[INSTANA_TRACE_ID_HEADER]
+          extract_instana_context(headers)
+        elsif headers['traceparent']
+          extract_w3c_context(headers)
+        else
+          {}
+        end
+      end
+
+      def extract_instana_context(headers)
         {
-          trace_id: headers['X-Instana-T'],
-          span_id: headers['X-Instana-S'],
-          level: headers['X-Instana-L']&.to_i
+          trace_id: headers[INSTANA_TRACE_ID_HEADER],
+          span_id: headers[INSTANA_SPAN_ID_HEADER],
+          level: headers[INSTANA_LEVEL_HEADER]&.to_i || 1
         }.reject { |_, v| v.nil? }
       end
 
+      def extract_w3c_context(headers)
+        m = headers['traceparent'].match(
+          /[0-9a-f][0-9a-e]-(?<trace>[0-9a-f]{32})-(?<parent>[0-9a-f]{16})-(?<flags>[0-9a-f]{2})/
+        )
+        return {} unless m
+
+        level = (m['flags'].to_i(16) & 0x01) == 1 ? 1 : 0
+        ts_match = headers['tracestate']&.match(/in=(?<t>[0-9a-f]+);(?<s>[0-9a-f]+)/)
+
+        if ts_match
+          { trace_id: ts_match['t'], span_id: ts_match['s'], level: level }
+        else
+          { trace_id: ::Instana::Util.header_to_id(m['trace'][16..]),
+            span_id: ::Instana::Util.header_to_id(m['parent']),
+            level: level }
+        end
+      end
+
       def log_error(error)
         # Log errors on to console if INSTANA_DEBUG is enabled
         ::Instana.logger.debug { "#{__method__}:#{File.basename(__FILE__)}:#{__LINE__}: #{error.message}" }

lib/instana/instrumentation/grpc.rb

@@ -8,8 +8,8 @@ module Instrumentation
     module GRPCCientInstrumentation
       CALL_TYPES = [:request_response, :client_streamer, :server_streamer, :bidi_streamer]
 
-      CALL_TYPES.each do |call_type|
-        define_method(call_type) do |method, *others, **options|
+      CALL_TYPES.each do |call_type| # rubocop:disable Metrics/BlockLength
+        define_method(call_type) do |method, *others, **options| # rubocop:disable Metrics/BlockLength
           begin
             kvs = { rpc: {} }
 
@@ -26,10 +26,15 @@ module GRPCCientInstrumentation
 
             context = ::Instana.tracer.context
             if context
-              options[:metadata] = (options[:metadata] || {}).merge(
+              extra = {
                 'x-instana-t' => context.trace_id_header,
-                'x-instana-s' => context.span_id_header
-              )
+                'x-instana-s' => context.span_id_header,
+                'x-instana-l' => context.level.to_s,
+                'traceparent' => context.trace_parent_header,
+                'tracestate' => context.trace_state_header
+              }
+              extra.reject! { |_, v| v.nil? || v.empty? }
+              options[:metadata] = (options[:metadata] || {}).merge(extra)
             end
 
             super(method, *others, **options)
@@ -58,12 +63,7 @@ module GRPCServerInstrumentation
             kvs = { rpc: {} }
             metadata = active_call.metadata
 
-            incoming_context = nil
-            if metadata.key?('x-instana-t')
-              incoming_context = SpanContext.new(trace_id: ::Instana::Util.header_to_id(metadata['x-instana-t']),
-                                                 span_id: metadata.key?('x-instana-s') ? ::Instana::Util.header_to_id(metadata['x-instana-s']) : nil,
-                                                 level: metadata.key?('x-instana-l') ? metadata['x-instana-l'] : nil)
-            end
+            incoming_context = extract_grpc_context(metadata)
 
             kvs[:rpc][:flavor] = :grpc
             kvs[:rpc][:host] = Socket.gethostname
@@ -87,6 +87,49 @@ module GRPCServerInstrumentation
           end
         end
       end
+
+      private
+
+      def extract_grpc_context(metadata)
+        if metadata.key?('x-instana-t')
+          extract_instana_grpc_context(metadata)
+        elsif metadata.key?('traceparent')
+          extract_w3c_grpc_context(metadata)
+        end
+      end
+
+      def extract_instana_grpc_context(metadata)
+        SpanContext.new(
+          trace_id: ::Instana::Util.header_to_id(metadata['x-instana-t']),
+          span_id: metadata.key?('x-instana-s') ? ::Instana::Util.header_to_id(metadata['x-instana-s']) : nil,
+          level: metadata['x-instana-l'] || 1
+        )
+      end
+
+      def extract_w3c_grpc_context(metadata)
+        m = metadata['traceparent'].match(
+          /[0-9a-f][0-9a-e]-(?<trace>[0-9a-f]{32})-(?<parent>[0-9a-f]{16})-(?<flags>[0-9a-f]{2})/
+        )
+        return nil unless m
+
+        level    = (m['flags'].to_i(16) & 0x01) == 1 ? 1 : 0
+        ts_match = metadata['tracestate']&.match(/in=(?<t>[0-9a-f]+);(?<s>[0-9a-f]+)/)
+        baggage  = {
+          external_trace_id: m['trace'],
+          external_trace_flags: m['flags'],
+          external_state: metadata['tracestate']
+        }
+
+        if ts_match
+          SpanContext.new(trace_id: ::Instana::Util.header_to_id(ts_match['t']),
+                          span_id: ::Instana::Util.header_to_id(ts_match['s']),
+                          level: level, baggage: baggage)
+        else
+          SpanContext.new(trace_id: ::Instana::Util.header_to_id(m['trace'][16..]),
+                          span_id: ::Instana::Util.header_to_id(m['parent']),
+                          level: level, baggage: baggage)
+        end
+      end
     end
   end
 end