Add ActivityPub attributes to current span when processing Activities (#40041)

app/controllers/activitypub/inboxes_controller.rb

@@ -4,6 +4,7 @@ class ActivityPub::InboxesController < ActivityPub::BaseController
   include JsonLdHelper
 
   before_action :skip_large_payload
+  before_action :decorate_trace
   before_action :skip_unknown_actor_activity
   before_action :require_actor_signature!
   skip_before_action :authenticate_user!
@@ -21,15 +22,19 @@ class ActivityPub::InboxesController < ActivityPub::BaseController
     head 413 if request.content_length > ActivityPub::Activity::MAX_JSON_SIZE
   end
 
+  def decorate_trace
+    ActivityPub::OpenTelemetry.decorate_current_span(payload: parsed_body)
+  end
+
   def skip_unknown_actor_activity
     head 202 if unknown_affected_account?
   end
 
   def unknown_affected_account?
-    json = JSON.parse(body)
+    json = parsed_body
+    return false if json.nil?
+
     json.is_a?(Hash) && %w(Delete Update).include?(json['type']) && json['actor'].present? && json['actor'] == value_or_id(json['object']) && !Account.exists?(uri: json['actor'])
-  rescue JSON::ParserError
-    false
   end
 
   def account_required?
@@ -55,6 +60,14 @@ class ActivityPub::InboxesController < ActivityPub::BaseController
     @body
   end
 
+  def parsed_body
+    return @parsed_body if defined?(@parsed_body)
+
+    @parsed_body = JSON.parse(body)
+  rescue JSON::ParserError
+    @parsed_body = nil
+  end
+
   def upgrade_account
     if signed_request_account&.ostatus?
       signed_request_account.update(last_webfingered_at: nil)

app/lib/activitypub/opentelemetry.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+module ActivityPub::OpenTelemetry
+  def self.decorate_current_span(payload:, namespace: 'activity')
+    decorate_span(span: OpenTelemetry::Trace.current_span, payload:, namespace:)
+  end
+
+  def self.decorate_span(span:, payload:, namespace: 'activity')
+    return unless span.recording?
+    return unless payload.is_a?(Hash)
+
+    id = payload['id']
+    type = payload['type']
+
+    span.set_attribute("activitypub.#{namespace}.id", id) if id.present?
+    span.set_attribute("activitypub.#{namespace}.type", type) if type.present?
+  end
+end

app/services/activitypub/process_activity_service.rb

@@ -11,6 +11,8 @@ class ActivityPub::ProcessActivityService < BaseService
 
     return unless @json.is_a?(Hash)
 
+    ActivityPub::OpenTelemetry.decorate_current_span(payload: @json)
+
     # Ideally, we should treat all ActivityPub payloads as proper JSON-LD.
     # However, some implementations do not produce valid JSON-LD, and processing JSON-LD is pretty expensive.
     # Therefore, we only process activities as JSON-LD if they make use of JSON-LD signatures.

config/initializers/inflections.rb

@@ -30,6 +30,7 @@ ActiveSupport::Inflector.inflections(:en) do |inflect|
   inflect.acronym 'SEO'
   inflect.acronym 'TOC'
   inflect.acronym 'URL'
+  inflect.acronym 'OpenTelemetry'
 
   inflect.singular 'data', 'data'
 end

spec/lib/activitypub/opentelemetry_spec.rb

@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+# rubocop:disable RSpec/SpecFilePathFormat
+# We write OpenTelemetry but files are like opentelemetry to
+# follow the main gem naming convention
+RSpec.describe ActivityPub::OpenTelemetry do
+  subject { described_class }
+
+  let(:payload) do
+    {
+      '@context' => 'https://www.w3.org/ns/activitystreams',
+      'id' => 'foo',
+      'type' => 'Create',
+    }
+  end
+
+  let(:span) { instance_double(OpenTelemetry::Trace::Span) }
+
+  before do
+    allow(span).to receive(:recording?).and_return(true)
+    allow(span).to receive(:set_attribute)
+  end
+
+  describe '#decorate_span' do
+    it 'adds attributes to span' do
+      subject.decorate_span(span:, payload:)
+
+      expect(span).to have_received(:set_attribute).with('activitypub.activity.id', payload['id'])
+      expect(span).to have_received(:set_attribute).with('activitypub.activity.type', payload['type'])
+    end
+
+    it 'can use different namespaces' do
+      subject.decorate_span(span:, payload:, namespace: 'object')
+
+      expect(span).to have_received(:set_attribute).with('activitypub.object.id', payload['id'])
+      expect(span).to have_received(:set_attribute).with('activitypub.object.type', payload['type'])
+    end
+  end
+
+  describe '#decorate_current_span' do
+    before do
+      allow(OpenTelemetry::Trace).to receive(:current_span).and_return(span)
+    end
+
+    it 'adds attributes to span' do
+      subject.decorate_current_span(payload:)
+
+      expect(span).to have_received(:set_attribute).with('activitypub.activity.id', payload['id'])
+      expect(span).to have_received(:set_attribute).with('activitypub.activity.type', payload['type'])
+    end
+  end
+end
+# rubocop:enable RSpec/SpecFilePathFormat

spec/requests/activitypub/inboxes_spec.rb

@@ -163,6 +163,30 @@ RSpec.describe 'ActivityPub Inboxes' do
           .to have_http_status(202)
       end
 
+      context 'with OpenTelemetry traces' do
+        subject { post inbox_path, params: activity.to_json, headers:, sign_with: remote_account }
+
+        let(:activity) { { id: 'https://example.net/~mallory/87374', type: 'Like' } }
+
+        let(:headers) { { 'CONTENT_TYPE' => 'application/json' } }
+
+        let(:span) { instance_double(OpenTelemetry::Trace::Span) }
+
+        before do
+          allow(OpenTelemetry::Trace).to receive(:current_span).and_return(span)
+          allow(span).to receive(:recording?).and_return(true)
+          allow(span).to receive(:set_attribute)
+        end
+
+        it 'adds attributes to current span' do
+          subject
+
+          expect(response).to have_http_status(202)
+          expect(span).to have_received(:set_attribute).with('activitypub.activity.id', activity[:id])
+          expect(span).to have_received(:set_attribute).with('activitypub.activity.type', activity[:type])
+        end
+      end
+
       def stub_follow_sync_worker
         allow(ActivityPub::FollowersSynchronizationWorker)
           .to receive(:perform_async)

spec/services/activitypub/process_activity_service_spec.rb

@@ -322,5 +322,22 @@ RSpec.describe ActivityPub::ProcessActivityService do
         end
       end
     end
+
+    context 'with OpenTelemetry traces' do
+      let(:span) { instance_double(OpenTelemetry::Trace::Span) }
+
+      before do
+        allow(OpenTelemetry::Trace).to receive(:current_span).and_return(span)
+        allow(span).to receive(:recording?).and_return(true)
+        allow(span).to receive(:set_attribute)
+      end
+
+      it 'adds attributes to current span' do
+        subject.call(json, actor)
+
+        expect(span).to have_received(:set_attribute).with('activitypub.activity.id', payload[:id])
+        expect(span).to have_received(:set_attribute).with('activitypub.activity.type', payload[:type])
+      end
+    end
   end
 end