Add Open Telemetry Spans throughout the codebase (#8488)

Co-authored-by: Jamie Magee <[email protected]>

sorbet/rbi/shims/opentelemetry-sdk.rbi

@@ -2,7 +2,85 @@
 # frozen_string_literal: true
 
 module OpenTelemetry
+  sig { returns(Trace::TracerProvider) }
+  def self.tracer_provider; end
+
   module SDK
-    def self.configure; end
+    sig do
+      params(
+        block: T.nilable(T.proc.params(arg0: Configurator).void)
+      )
+        .void
+    end
+    def self.configure(&block); end
+
+    class Configurator
+      sig { params(service_name: String, config: T.nilable(T::Hash[String, T.untyped])).void }
+      def service_name=(service_name, config = nil); end
+
+      sig { params(instrumentation_name: String).void }
+      def use(instrumentation_name); end
+    end
+  end
+
+  module Trace
+    def self.current_span; end
+
+    module Status
+      sig { params(message: String).void }
+      def self.error(message); end
+    end
+
+    module Tracer
+      sig do
+        params(
+          name: String,
+          attributes: T.nilable(T::Hash[String, T.untyped]),
+          links: T.nilable(T::Array[Link]),
+          start_timestamp: T.nilable(Integer),
+          kind: T.nilable(Symbol),
+          block: T.nilable(T.proc.params(arg0: Span, arg1: Context).void)
+        )
+          .void
+      end
+      def in_span(name, attributes: nil, links: nil, start_timestamp: nil, kind: nil, &block); end
+
+      sig do
+        params(
+          name: String,
+          with_parent: T.nilable(Span),
+          attributes: T.nilable(T::Hash[String, T.untyped]),
+          links: T.nilable(T::Array[Link]),
+          start_timestamp: T.nilable(Integer),
+          kind: T.nilable(Symbol)
+        )
+          .returns(Span)
+      end
+      def start_span(name, with_parent: nil, attributes: nil, links: nil, start_timestamp: nil, kind: nil); end
+    end
+
+    class TracerProvider
+      sig { params(name: T.nilable(String), version: T.nilable(String)).returns(Tracer) }
+      def tracer(name = nil, version = nil); end
+    end
+
+    class Link; end
+
+    class Span
+      sig do
+        params(
+          key: String,
+          value: T.untyped
+        )
+          .returns(T.self_type)
+      end
+      def set_attribute(key, value); end
+      sig { void }
+      def finish; end
+    end
+  end
+
+  class Context
+    class Key; end
   end
 end

updater/lib/dependabot/api_client.rb

@@ -3,6 +3,7 @@
 
 require "http"
 require "dependabot/job"
+require "dependabot/opentelemetry"
 require "sorbet-runtime"
 
 # Provides a client to access the internal Dependabot Service's API
@@ -30,6 +31,11 @@ def initialize(base_url, job_id, job_token)
     # TODO: Make `base_commit_sha` part of Dependabot::DependencyChange
     sig { params(dependency_change: Dependabot::DependencyChange, base_commit_sha: String).void }
     def create_pull_request(dependency_change, base_commit_sha)
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("create_pull_request", kind: :internal)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::JOB_ID, job_id)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::BASE_COMMIT_SHA, base_commit_sha)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::DEPENDENCY_NAMES, dependency_change.humanized)
+
       api_url = "#{base_url}/update_jobs/#{job_id}/create_pull_request"
       data = create_pull_request_data(dependency_change, base_commit_sha)
       response = http_client.post(api_url, json: { data: data })
@@ -41,12 +47,19 @@ def create_pull_request(dependency_change, base_commit_sha)
 
       sleep(rand(3.0..10.0))
       retry
+    ensure
+      span&.finish
     end
 
     # TODO: Make `base_commit_sha` part of Dependabot::DependencyChange
     # TODO: Determine if we should regenerate the PR message within core for updates
     sig { params(dependency_change: Dependabot::DependencyChange, base_commit_sha: String).void }
     def update_pull_request(dependency_change, base_commit_sha)
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("update_pull_request", kind: :internal)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::JOB_ID, job_id)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::BASE_COMMIT_SHA, base_commit_sha)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::DEPENDENCY_NAMES, dependency_change.humanized)
+
       api_url = "#{base_url}/update_jobs/#{job_id}/update_pull_request"
       body = {
         data: {
@@ -64,10 +77,16 @@ def update_pull_request(dependency_change, base_commit_sha)
 
       sleep(rand(3.0..10.0))
       retry
+    ensure
+      span&.finish
     end
 
     sig { params(dependency_names: T.any(String, T::Array[String]), reason: T.any(String, Symbol)).void }
     def close_pull_request(dependency_names, reason)
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("close_pull_request", kind: :internal)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::JOB_ID, job_id)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::PR_CLOSE_REASON, reason)
+
       api_url = "#{base_url}/update_jobs/#{job_id}/close_pull_request"
       body = { data: { "dependency-names": dependency_names, reason: reason } }
       response = http_client.post(api_url, json: body)
@@ -79,10 +98,15 @@ def close_pull_request(dependency_names, reason)
 
       sleep(rand(3.0..10.0))
       retry
+    ensure
+      span&.finish
     end
 
     sig { params(error_type: T.any(String, Symbol), error_details: T.nilable(T::Hash[T.untyped, T.untyped])).void }
     def record_update_job_error(error_type:, error_details:)
+      ::Dependabot::OpenTelemetry.record_update_job_error(job_id: job_id, error_type: error_type,
+                                                          error_details: error_details)
+
       api_url = "#{base_url}/update_jobs/#{job_id}/record_update_job_error"
       body = {
         data: {
@@ -104,6 +128,8 @@ def record_update_job_error(error_type:, error_details:)
     sig { params(error_type: T.any(Symbol, String), error_details: T.nilable(T::Hash[T.untyped, T.untyped])).void }
     def record_update_job_unknown_error(error_type:, error_details:)
       error_type = "unknown_error" if error_type.nil?
+      ::Dependabot::OpenTelemetry.record_update_job_error(job_id: job_id, error_type: error_type,
+                                                          error_details: error_details)
 
       api_url = "#{base_url}/update_jobs/#{job_id}/record_update_job_unknown_error"
       body = {
@@ -125,6 +151,10 @@ def record_update_job_unknown_error(error_type:, error_details:)
 
     sig { params(base_commit_sha: String).void }
     def mark_job_as_processed(base_commit_sha)
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("mark_job_as_processed", kind: :internal)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::BASE_COMMIT_SHA, base_commit_sha)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::JOB_ID, job_id)
+
       api_url = "#{base_url}/update_jobs/#{job_id}/mark_as_processed"
       body = { data: { "base-commit-sha": base_commit_sha } }
       response = http_client.patch(api_url, json: body)
@@ -136,10 +166,15 @@ def mark_job_as_processed(base_commit_sha)
 
       sleep(rand(3.0..10.0))
       retry
+    ensure
+      span&.finish
     end
 
     sig { params(dependencies: T::Array[T::Hash[Symbol, T.untyped]], dependency_files: T::Array[DependencyFile]).void }
     def update_dependency_list(dependencies, dependency_files)
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("update_dependency_list", kind: :internal)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::JOB_ID, job_id)
+
       api_url = "#{base_url}/update_jobs/#{job_id}/update_dependency_list"
       body = {
         data: {
@@ -156,6 +191,8 @@ def update_dependency_list(dependencies, dependency_files)
 
       sleep(rand(3.0..10.0))
       retry
+    ensure
+      span&.finish
     end
 
     sig { params(ecosystem_versions: T::Hash[Symbol, T.untyped]).void }
@@ -177,6 +214,13 @@ def record_ecosystem_versions(ecosystem_versions)
 
     sig { params(metric: String, tags: T::Hash[String, String]).void }
     def increment_metric(metric, tags:)
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("increment_metric", kind: :internal)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::JOB_ID, job_id)
+      span&.set_attribute(::Dependabot::OpenTelemetry::Attributes::METRIC, metric)
+      tags.each do |key, value|
+        span&.set_attribute(key, value)
+      end
+
       api_url = "#{base_url}/update_jobs/#{job_id}/increment_metric"
       body = {
         data: {
@@ -189,6 +233,8 @@ def increment_metric(metric, tags:)
       Dependabot.logger.debug("Unable to report metric '#{metric}'.") if response.code >= 400
     rescue HTTP::ConnectionError, OpenSSL::SSL::SSLError
       Dependabot.logger.debug("Unable to report metric '#{metric}'.")
+    ensure
+      span&.finish
     end
 
     private

updater/lib/dependabot/file_fetcher_command.rb

@@ -3,6 +3,7 @@
 
 require "base64"
 require "dependabot/base_command"
+require "dependabot/opentelemetry"
 require "dependabot/updater"
 require "octokit"
 
@@ -13,9 +14,11 @@ class FileFetcherCommand < BaseCommand
     # NotImplementedError if it is referenced
     attr_reader :base_commit_sha
 
-    def perform_job
+    def perform_job # rubocop:disable Metrics/PerceivedComplexity
       @base_commit_sha = nil
 
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("perform_job", kind: :internal)
+
       begin
         connectivity_check if ENV["ENABLE_CONNECTIVITY_CHECK"] == "1"
         clone_repo_contents
@@ -49,6 +52,8 @@ def perform_job
                                           ))
 
       save_job_details
+    ensure
+      span&.finish
     end
 
     private

updater/lib/dependabot/opentelemetry.rb

@@ -2,11 +2,22 @@
 # frozen_string_literal: true
 
 require "sorbet-runtime"
+require "opentelemetry/sdk"
 
 module Dependabot
   module OpenTelemetry
     extend T::Sig
 
+    module Attributes
+      JOB_ID = "dependabot.job.id"
+      ERROR_TYPE = "dependabot.job.error_type"
+      ERROR_DETAILS = "dependabot.job.error_details"
+      METRIC = "dependabot.metric"
+      BASE_COMMIT_SHA = "dependabot.base_commit_sha"
+      DEPENDENCY_NAMES = "dependabot.dependency_names"
+      PR_CLOSE_REASON = "dependabot.pr_close_reason"
+    end
+
     sig { returns(T::Boolean) }
     def self.should_configure?
       ENV["OTEL_ENABLED"] == "true"
@@ -16,7 +27,8 @@ def self.should_configure?
     def self.configure
       return unless should_configure?
 
-      require "opentelemetry/sdk"
+      puts "OpenTelemetry is enabled, configuring..."
+
       require "opentelemetry/exporter/otlp"
       require "opentelemetry/instrumentation/excon"
       require "opentelemetry/instrumentation/faraday"
@@ -28,6 +40,58 @@ def self.configure
         config.use "OpenTelemetry::Instrumentation::Faraday"
         config.use "OpenTelemetry::Instrumentation::Http"
       end
+
+      tracer
+    end
+
+    sig { returns(T.nilable(::OpenTelemetry::Trace::Tracer)) }
+    def self.tracer
+      return unless should_configure?
+
+      ::OpenTelemetry.tracer_provider.tracer("dependabot", Dependabot::VERSION)
+    end
+
+    sig do
+      params(
+        job_id: T.any(String, Integer),
+        error_type: T.any(String, Symbol),
+        error_details: T.nilable(T::Hash[T.untyped, T.untyped])
+      ).void
+    end
+    def self.record_update_job_error(job_id:, error_type:, error_details:)
+      return unless should_configure?
+
+      current_span = ::OpenTelemetry::Trace.current_span
+
+      attributes = {
+        Attributes::JOB_ID => job_id,
+        Attributes::ERROR_TYPE => error_type
+      }
+
+      error_details&.each do |key, value|
+        attributes.store("#{Attributes::ERROR_DETAILS}.#{key}", value)
+      end
+
+      current_span.add_event(error_type, attributes: attributes)
+    end
+
+    sig do
+      params(
+        error: StandardError,
+        job: T.untyped,
+        tags: T::Hash[String, T.untyped]
+      ).void
+    end
+    def self.record_exception(error:, job: nil, tags: {})
+      return unless should_configure?
+
+      current_span = ::OpenTelemetry::Trace.current_span
+
+      current_span.set_attribute(Attributes::JOB_ID, job.id) if job
+      current_span.add_attributes(tags) if tags.any?
+
+      current_span.status = ::OpenTelemetry::Trace::Status.error(error.message)
+      current_span.record_exception(error)
     end
   end
 end

updater/lib/dependabot/service.rb

@@ -4,6 +4,7 @@
 require "raven"
 require "terminal-table"
 require "dependabot/api_client"
+require "dependabot/opentelemetry"
 require "sorbet-runtime"
 
 # This class provides an output adapter for the Dependabot Service which manages
@@ -98,6 +99,7 @@ def update_dependency_list(dependency_snapshot:)
       ).void
     end
     def capture_exception(error:, job: nil, dependency: nil, dependency_group: nil, tags: {}, extra: {})
+      ::Dependabot::OpenTelemetry.record_exception(error: error, job: job, tags: tags)
       T.unsafe(Raven).capture_exception(
         error,
         {

updater/lib/dependabot/update_files_command.rb

@@ -4,6 +4,7 @@
 require "base64"
 require "dependabot/base_command"
 require "dependabot/dependency_snapshot"
+require "dependabot/opentelemetry"
 require "dependabot/updater"
 
 module Dependabot
@@ -13,6 +14,7 @@ def perform_job
       # encoded files and commit information in the environment, so let's retrieve
       # them, decode and parse them into an object that knows the current state
       # of the project's dependencies.
+      span = ::Dependabot::OpenTelemetry.tracer&.start_span("perform_job", kind: :internal)
       begin
         dependency_snapshot = Dependabot::DependencySnapshot.create_from_job_definition(
           job: job,
@@ -42,6 +44,8 @@ def perform_job
       # reported errors to the service, but we always consider the job as
       # successfully processed unless it actually raises.
       service.mark_job_as_processed(dependency_snapshot.base_commit_sha)
+    ensure
+      span&.finish
     end
 
     private