Merge pull request #189 from DataDog/anmarchenko/code_coverage_multi_…

…threading_support

[SDTEST-408] multi threaded code coverage support for datadog_cov

ext/datadog_cov/datadog_cov.c

@@ -1,7 +1,11 @@
 #include <ruby.h>
 #include <ruby/debug.h>
 
-const int PROFILE_FRAMES_BUFFER_SIZE = 1;
+#define PROFILE_FRAMES_BUFFER_SIZE 1
+
+// threading modes
+#define SINGLE_THREADED_COVERAGE_MODE 0
+#define MULTI_THREADED_COVERAGE_MODE 1
 
 char *ruby_strndup(const char *str, size_t size)
 {
@@ -26,12 +30,18 @@ struct dd_cov_data
   VALUE coverage;
 
   uintptr_t last_filename_ptr;
+
+  // for single threaded mode: thread that is being covered
+  VALUE th_covered;
+
+  int threading_mode;
 };
 
 static void dd_cov_mark(void *ptr)
 {
   struct dd_cov_data *dd_cov_data = ptr;
   rb_gc_mark_movable(dd_cov_data->coverage);
+  rb_gc_mark_movable(dd_cov_data->th_covered);
 }
 
 static void dd_cov_free(void *ptr)
@@ -46,6 +56,7 @@ static void dd_cov_compact(void *ptr)
 {
   struct dd_cov_data *dd_cov_data = ptr;
   dd_cov_data->coverage = rb_gc_location(dd_cov_data->coverage);
+  dd_cov_data->th_covered = rb_gc_location(dd_cov_data->th_covered);
 }
 
 const rb_data_type_t dd_cov_data_type = {
@@ -68,6 +79,7 @@ static VALUE dd_cov_allocate(VALUE klass)
   dd_cov_data->ignored_path = NULL;
   dd_cov_data->ignored_path_len = 0;
   dd_cov_data->last_filename_ptr = 0;
+  dd_cov_data->threading_mode = MULTI_THREADED_COVERAGE_MODE;
 
   return obj;
 }
@@ -85,9 +97,25 @@ static VALUE dd_cov_initialize(int argc, VALUE *argv, VALUE self)
   }
   VALUE rb_ignored_path = rb_hash_lookup(opt, ID2SYM(rb_intern("ignored_path")));
 
+  VALUE rb_threading_mode = rb_hash_lookup(opt, ID2SYM(rb_intern("threading_mode")));
+  int threading_mode;
+  if (rb_threading_mode == ID2SYM(rb_intern("multi")))
+  {
+    threading_mode = MULTI_THREADED_COVERAGE_MODE;
+  }
+  else if (rb_threading_mode == ID2SYM(rb_intern("single")))
+  {
+    threading_mode = SINGLE_THREADED_COVERAGE_MODE;
+  }
+  else
+  {
+    rb_raise(rb_eArgError, "threading mode is invalid");
+  }
+
   struct dd_cov_data *dd_cov_data;
   TypedData_Get_Struct(self, struct dd_cov_data, &dd_cov_data_type, dd_cov_data);
 
+  dd_cov_data->threading_mode = threading_mode;
   dd_cov_data->root_len = RSTRING_LEN(rb_root);
   dd_cov_data->root = ruby_strndup(RSTRING_PTR(rb_root), dd_cov_data->root_len);
 
@@ -152,8 +180,6 @@ static void dd_cov_update_coverage(rb_event_flag_t event, VALUE data, VALUE self
 
 static VALUE dd_cov_start(VALUE self)
 {
-  // get current thread
-  VALUE thval = rb_thread_current();
 
   struct dd_cov_data *dd_cov_data;
   TypedData_Get_Struct(self, struct dd_cov_data, &dd_cov_data_type, dd_cov_data);
@@ -163,22 +189,41 @@ static VALUE dd_cov_start(VALUE self)
     rb_raise(rb_eRuntimeError, "root is required");
   }
 
-  // add event hook
-  rb_thread_add_event_hook(thval, dd_cov_update_coverage, RUBY_EVENT_LINE, self);
+  if (dd_cov_data->threading_mode == SINGLE_THREADED_COVERAGE_MODE)
+  {
+    VALUE thval = rb_thread_current();
+    rb_thread_add_event_hook(thval, dd_cov_update_coverage, RUBY_EVENT_LINE, self);
+    dd_cov_data->th_covered = thval;
+  }
+  else
+  {
+    rb_add_event_hook(dd_cov_update_coverage, RUBY_EVENT_LINE, self);
+  }
 
   return self;
 }
 
 static VALUE dd_cov_stop(VALUE self)
 {
-  // get current thread
-  VALUE thval = rb_thread_current();
-  // remove event hook for the current thread
-  rb_thread_remove_event_hook(thval, dd_cov_update_coverage);
-
   struct dd_cov_data *dd_cov_data;
   TypedData_Get_Struct(self, struct dd_cov_data, &dd_cov_data_type, dd_cov_data);
 
+  if (dd_cov_data->threading_mode == SINGLE_THREADED_COVERAGE_MODE)
+  {
+    VALUE thval = rb_thread_current();
+    if (!rb_equal(thval, dd_cov_data->th_covered))
+    {
+      rb_raise(rb_eRuntimeError, "Coverage was not started by this thread");
+    }
+
+    rb_thread_remove_event_hook(dd_cov_data->th_covered, dd_cov_update_coverage);
+    dd_cov_data->th_covered = Qnil;
+  }
+  else
+  {
+    rb_remove_event_hook(dd_cov_update_coverage);
+  }
+
   VALUE res = dd_cov_data->coverage;
 
   dd_cov_data->coverage = rb_hash_new();

lib/datadog/ci/configuration/components.rb

@@ -129,7 +129,8 @@ def activate_ci!(settings)
             config_tags: custom_configuration_tags,
             coverage_writer: coverage_writer,
             enabled: settings.ci.enabled && settings.ci.itr_enabled,
-            bundle_location: settings.ci.itr_code_coverage_excluded_bundle_path
+            bundle_location: settings.ci.itr_code_coverage_excluded_bundle_path,
+            use_single_threaded_coverage: settings.ci.itr_code_coverage_use_single_threaded_mode
           )
 
           git_tree_uploader = Git::TreeUploader.new(api: test_visibility_api)

lib/datadog/ci/configuration/settings.rb

@@ -76,6 +76,12 @@ def self.add_settings!(base)
                 end
               end
 
+              option :itr_code_coverage_use_single_threaded_mode do |o|
+                o.type :bool
+                o.env CI::Ext::Settings::ENV_ITR_CODE_COVERAGE_USE_SINGLE_THREADED_MODE
+                o.default false
+              end
+
               define_method(:instrument) do |integration_name, options = {}, &block|
                 return unless enabled
 

lib/datadog/ci/ext/settings.rb

@@ -13,6 +13,7 @@ module Settings
         ENV_ITR_ENABLED = "DD_CIVISIBILITY_ITR_ENABLED"
         ENV_GIT_METADATA_UPLOAD_ENABLED = "DD_CIVISIBILITY_GIT_METADATA_UPLOAD_ENABLED"
         ENV_ITR_CODE_COVERAGE_EXCLUDED_BUNDLE_PATH = "DD_CIVISIBILITY_ITR_CODE_COVERAGE_EXCLUDED_BUNDLE_PATH"
+        ENV_ITR_CODE_COVERAGE_USE_SINGLE_THREADED_MODE = "DD_CIVISIBILITY_ITR_CODE_COVERAGE_USE_SINGLE_THREADED_MODE"
 
         # Source: https://docs.datadoghq.com/getting_started/site/
         DD_SITE_ALLOWLIST = %w[

lib/datadog/ci/itr/runner.rb

@@ -31,7 +31,8 @@ def initialize(
           api: nil,
           coverage_writer: nil,
           enabled: false,
-          bundle_location: nil
+          bundle_location: nil,
+          use_single_threaded_coverage: false
         )
           @enabled = enabled
           @api = api
@@ -43,6 +44,7 @@ def initialize(
           else
             bundle_location
           end
+          @use_single_threaded_coverage = use_single_threaded_coverage
 
           @test_skipping_enabled = false
           @code_coverage_enabled = false
@@ -186,12 +188,15 @@ def write(event)
         def coverage_collector
           Thread.current[:dd_coverage_collector] ||= Coverage::DDCov.new(
             root: Git::LocalRepository.root,
-            ignored_path: @bundle_location
+            ignored_path: @bundle_location,
+            threading_mode: code_coverage_mode
           )
         end
 
         def load_datadog_cov!
           require "datadog_cov.#{RUBY_VERSION}_#{RUBY_PLATFORM}"
+
+          Datadog.logger.debug("Loaded Datadog code coverage collector, using coverage mode: #{code_coverage_mode}")
         rescue LoadError => e
           Datadog.logger.error("Failed to load coverage collector: #{e}. Code coverage will not be collected.")
 
@@ -222,6 +227,10 @@ def fetch_skippable_tests(test_session:, git_tree_upload_worker:)
           Datadog.logger.debug { "Found #{@skippable_tests.count} skippable tests." }
           Datadog.logger.debug { "ITR correlation ID: #{@correlation_id}" }
         end
+
+        def code_coverage_mode
+          @use_single_threaded_coverage ? :single : :multi
+        end
       end
     end
   end

sig/datadog/ci/ext/settings.rbs

@@ -10,6 +10,7 @@ module Datadog
         ENV_ITR_ENABLED: String
         ENV_GIT_METADATA_UPLOAD_ENABLED: String
         ENV_ITR_CODE_COVERAGE_EXCLUDED_BUNDLE_PATH: String
+        ENV_ITR_CODE_COVERAGE_USE_SINGLE_THREADED_MODE: String
 
         DD_SITE_ALLOWLIST: Array[String]
       end

sig/datadog/ci/itr/coverage/ddcov.rbs

@@ -3,8 +3,9 @@ module Datadog
     module ITR
       module Coverage
         class DDCov
+          type threading_mode = :multi | :single
 
-          def initialize: (root: String, ignored_path: String?) -> void
+          def initialize: (root: String, ignored_path: String?, threading_mode: threading_mode) -> void
 
           def start: () -> void
 

sig/datadog/ci/itr/runner.rbs

@@ -15,6 +15,7 @@ module Datadog
         @dd_env: String?
         @config_tags: Hash[String, String]
         @bundle_location: String?
+        @use_single_threaded_coverage: bool
 
         @skipped_tests_count: Integer
         @mutex: Thread::Mutex
@@ -23,7 +24,7 @@ module Datadog
         attr_reader skipped_tests_count: Integer
         attr_reader correlation_id: String?
 
-        def initialize: (dd_env: String?, ?enabled: bool, ?coverage_writer: Datadog::CI::ITR::Coverage::Writer?, ?api: Datadog::CI::Transport::Api::Base?, ?config_tags: Hash[String, String]?, ?bundle_location: String?) -> void
+        def initialize: (dd_env: String?, ?enabled: bool, ?coverage_writer: Datadog::CI::ITR::Coverage::Writer?, ?api: Datadog::CI::Transport::Api::Base?, ?config_tags: Hash[String, String]?, ?bundle_location: String?, ?use_single_threaded_coverage: bool) -> void
 
         def configure: (Hash[String, untyped] remote_configuration, test_session: Datadog::CI::TestSession, git_tree_upload_worker: Datadog::CI::Worker) -> void
 
@@ -58,6 +59,8 @@ module Datadog
         def fetch_skippable_tests: (test_session: Datadog::CI::TestSession, git_tree_upload_worker: Datadog::CI::Worker) -> void
 
         def increment_skipped_tests_counter: () -> void
+
+        def code_coverage_mode: () -> Datadog::CI::ITR::Coverage::DDCov::threading_mode
       end
     end
   end

spec/datadog/ci/contrib/minitest/helpers/addition_helper.rb

@@ -0,0 +1,5 @@
+module AdditionHelper
+  def self.add(a, b)
+    a + b
+  end
+end

spec/datadog/ci/contrib/minitest/instrumentation_spec.rb

@@ -27,7 +27,7 @@ def test_foo
     end
   end
 
-  context "with service name configured" do
+  context "with service name configured and code coverage enabled" do
     include_context "CI mode activated" do
       let(:integration_name) { :minitest }
       let(:integration_options) { {service_name: "ltest"} }
@@ -402,12 +402,18 @@ def test_foo
         before(:context) do
           Minitest::Runnable.reset
 
+          require_relative "helpers/addition_helper"
           class SomeTest < Minitest::Test
             def test_pass
               assert true
             end
 
             def test_pass_other
+              # add thread to test that code coverage is collected
+              t = Thread.new do
+                AdditionHelper.add(1, 2)
+              end
+              t.join
               assert true
             end
           end
@@ -494,6 +500,13 @@ def test_pass_other
           expect_coverage_events_belong_to_suite(first_test_suite_span)
           expect_coverage_events_belong_to_tests(test_spans)
           expect_non_empty_coverages
+
+          # expect that background thread is covered
+          test_span = test_spans.find { |span| span.get_tag("test.name") == "test_pass_other" }
+          cov_event = find_coverage_for_test(test_span)
+          expect(cov_event.coverage.keys).to include(
+            File.expand_path(File.join(__dir__, "helpers/addition_helper.rb"))
+          )
         end
 
         context "when ITR skips tests" do
@@ -881,4 +894,49 @@ class SomeUnskippableSpec < Minitest::Spec
       end
     end
   end
+
+  context "when using single threaded code coverage" do
+    include_context "CI mode activated" do
+      let(:integration_name) { :minitest }
+
+      let(:itr_enabled) { true }
+      let(:code_coverage_enabled) { true }
+      let(:use_single_threaded_coverage) { true }
+    end
+
+    before do
+      Minitest.run([])
+    end
+
+    before(:context) do
+      Thread.current[:dd_coverage_collector] = nil
+
+      Minitest::Runnable.reset
+
+      require_relative "helpers/addition_helper"
+      class SomeTestWithThreads < Minitest::Test
+        def test_with_background_thread
+          # add thread to test that code coverage is collected
+          t = Thread.new do
+            AdditionHelper.add(1, 2)
+          end
+          t.join
+          assert true
+        end
+      end
+    end
+
+    it "does not cover the background thread" do
+      skip if PlatformHelpers.jruby?
+
+      expect(test_spans).to have(1).item
+      expect(coverage_events).to have(1).item
+
+      # expect that background thread is not covered
+      cov_event = find_coverage_for_test(first_test_span)
+      expect(cov_event.coverage.keys).not_to include(
+        File.expand_path(File.join(__dir__, "helpers/addition_helper.rb"))
+      )
+    end
+  end
 end

spec/datadog/ci/worker_spec.rb

@@ -24,7 +24,7 @@
 
     context "when the worker has started" do
       context "when the worker is running" do
-        let(:queue) { Queue.new }
+        let(:queue) { Thread::Queue.new }
         subject(:worker) { described_class.new { queue.pop } }
 
         it do