Merge pull request #593 from DataDog/anmarchenko/isolate-parallel-test-state

fix: isolate state between parallel_tests sessions

Author: anmarchenko

AGENTS.md

@@ -106,6 +106,7 @@ The native extension is in `ext/` directory and compiled for each Ruby version.
 
 - We use Ruby 2.7 syntax
 - Use exceptions for exceptional cases, not for control flow.
+- Instrumentation must never raise its own errors into customer test processes. Log internal failures at warn or error level and degrade gracefully instead.
 - Implement proper error logging and user-friendly messages.
 - Do not use `instance_variable_set` or `instance_variable_get`; add explicit APIs instead.
 
@@ -130,6 +131,7 @@ The native extension is in `ext/` directory and compiled for each Ruby version.
 
 - All changes should be covered by corresponding tests
 - Place tests in the `spec/datadog/ci` folder
+- Never monkey-patch production code in tests. Do not use techniques such as `prepend`, reopening production classes or modules, or replacing production singleton methods; exercise production code through its normal interfaces instead.
 - Do not use `instance_variable_set` or `instance_variable_get` in tests; use mocking or explicit APIs when needed
 - Do not make changes to `release_gem_spec.rb` if not asked
 - Do not use focused tests feature (fit, fdescribe)

lib/datadog/ci/contrib/parallel_tests/cli.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require_relative "../../ext/test"
+require_relative "../../utils/file_storage"
 require_relative "../rspec/ext"
 
 module Datadog
@@ -17,7 +18,7 @@ def run_tests_in_parallel(num_processes, options)
               # only rspec runner is supported for now
               return super if @runner != ::ParallelTests::RSpec::Runner
 
-              begin
+              Utils::FileStorage.with_new_namespace do |storage_namespace|
                 # Preserve activation explicitly for the RSpec worker commands.
                 # Starting the distributed parent session removes inherited
                 # activation before unrelated child processes can receive it.
@@ -35,6 +36,7 @@ def run_tests_in_parallel(num_processes, options)
 
                 options[:env] ||= {}
                 options[:env][CI::Ext::Settings::ENV_TEST_VISIBILITY_DRB_SERVER_URI] = test_tracing_component.context_service_uri
+                options[:env][Utils::FileStorage::ENV_NAMESPACE] = storage_namespace
                 options[:env]["RUBYOPT"] ||= worker_rubyopt if worker_rubyopt
 
                 super

lib/datadog/ci/test_session.rb

@@ -14,6 +14,12 @@ module CI
     class TestSession < ConcurrentSpan
       attr_accessor :estimated_total_tests_count, :distributed
 
+      def initialize(tracer_span)
+        super
+
+        @distributed = false
+      end
+
       # Finishes the current test session.
       # @return [void]
       def finish

lib/datadog/ci/test_tracing/component.rb

@@ -82,7 +82,7 @@ def configure(library_configuration, test_session)
           store_component_state if test_session.distributed
         end
 
-        def start_test_session(service: nil, tags: {}, estimated_total_tests_count: 0, distributed: false, local_test_suites_mode: true)
+        def start_test_session(service: nil, tags: {}, estimated_total_tests_count: 0, distributed: nil, local_test_suites_mode: true)
           return skip_tracing unless test_suite_level_visibility_enabled
 
           @local_test_suites_mode = local_test_suites_mode
@@ -91,7 +91,7 @@ def start_test_session(service: nil, tags: {}, estimated_total_tests_count: 0, d
 
           test_session = maybe_remote_context.start_test_session(service: service, tags: tags)
           test_session.estimated_total_tests_count = estimated_total_tests_count
-          test_session.distributed = distributed
+          test_session.distributed = distributed unless distributed.nil?
 
           on_test_session_started(test_session)
 
@@ -321,8 +321,6 @@ def on_test_session_finished(test_session)
           DeprecatedTotalCoverageMetric.extract_lines_pct(test_session)
 
           Telemetry.event_finished(test_session)
-
-          Utils::FileStorage.cleanup
         end
 
         def on_test_module_finished(test_module)

lib/datadog/ci/test_tracing/null_component.rb

@@ -9,7 +9,7 @@ def configure(_, _)
         end
 
         def start_test_session(
-          service: nil, tags: {}, estimated_total_tests_count: 0, distributed: false, local_test_suites_mode: true
+          service: nil, tags: {}, estimated_total_tests_count: 0, distributed: nil, local_test_suites_mode: true
         )
           skip_tracing
         end

lib/datadog/ci/utils/file_storage.rb

@@ -1,26 +1,39 @@
 # frozen_string_literal: true
 
 require "fileutils"
-require "tempfile"
+require "securerandom"
+require "tmpdir"
 
 module Datadog
   module CI
     module Utils
       # FileStorage module provides functionality for storing and retrieving arbitrary Ruby objects in a temp file
       # to share them between processes.
       module FileStorage
+        class MissingNamespaceError < StandardError; end
+
         TEMP_DIR = File.join(Dir.tmpdir, "datadog-ci-storage")
+        ENV_NAMESPACE = "DD_CIVISIBILITY_PARALLEL_TESTS_RUN_ID"
 
         def self.store(key, value)
           ensure_temp_dir_exists
           file_path = file_path_for(key)
+          temporary_path = File.join(storage_dir, "dd-ci-#{SecureRandom.uuid}.tmp")
 
-          File.binwrite(file_path, Marshal.dump(value))
+          File.open(temporary_path, File::WRONLY | File::CREAT | File::EXCL, 0o600) do |file|
+            file.binmode
+            file.write(Marshal.dump(value))
+            file.flush
+            file.fsync
+          end
+          File.rename(temporary_path, file_path)
 
           true
         rescue => e
-          Datadog.logger.error("Failed to store data for key '#{key}': #{e.class} - #{e.message}")
+          Datadog.logger.error("Failed to store data for key '#{key}': #{e.class}")
           false
+        ensure
+          FileUtils.rm_f(temporary_path) if temporary_path
         end
 
         def self.retrieve(key)
@@ -29,27 +42,54 @@ def self.retrieve(key)
 
           Marshal.load(File.binread(file_path))
         rescue => e
-          Datadog.logger.error("Failed to retrieve data for key '#{key}': #{e.class} - #{e.message}")
+          Datadog.logger.error("Failed to retrieve data for key '#{key}': #{e.class}")
           nil
         end
 
-        def self.cleanup
-          return false unless Dir.exist?(TEMP_DIR)
+        def self.cleanup(namespace)
+          directory = storage_dir(namespace)
+          return false unless Dir.exist?(directory)
 
-          FileUtils.rm_rf(TEMP_DIR)
+          FileUtils.rm_rf(directory)
           true
         rescue => e
-          Datadog.logger.error("Failed to cleanup storage directory: #{e.class} - #{e.message}")
+          Datadog.logger.error("Failed to cleanup storage directory: #{e.class}")
           false
         end
+        private_class_method :cleanup
+
+        def self.with_new_namespace
+          previous_namespace = ENV[ENV_NAMESPACE]
+          namespace = SecureRandom.uuid
+          ENV[ENV_NAMESPACE] = namespace
+
+          yield namespace
+        ensure
+          cleanup(namespace) if namespace
+
+          if previous_namespace
+            ENV[ENV_NAMESPACE] = previous_namespace
+          else
+            ENV.delete(ENV_NAMESPACE)
+          end
+        end
 
         def self.ensure_temp_dir_exists
-          FileUtils.mkdir_p(TEMP_DIR) unless Dir.exist?(TEMP_DIR)
+          FileUtils.mkdir_p(storage_dir)
         end
 
         def self.file_path_for(key)
           sanitized_key = key.to_s.gsub(/[^a-zA-Z0-9_-]/, "_")
-          File.join(TEMP_DIR, "dd-ci-#{sanitized_key}.dat")
+          File.join(storage_dir, "dd-ci-#{sanitized_key}.dat")
+        end
+
+        def self.storage_dir(namespace = ENV[ENV_NAMESPACE])
+          if namespace.nil? || namespace.empty?
+            raise MissingNamespaceError, "File storage namespace is not set"
+          end
+
+          sanitized_namespace = namespace.gsub(/[^a-zA-Z0-9_-]/, "_")
+          File.join(TEMP_DIR, sanitized_namespace)
         end
       end
     end

sig/datadog/ci/test_tracing/component.rbs

@@ -43,7 +43,7 @@ module Datadog
 
         def trace: (String span_name, ?type: String, ?tags: Hash[Symbol | String, untyped]) ?{ (Datadog::CI::Span span) -> untyped } -> untyped
 
-        def start_test_session: (?service: String?, ?tags: Hash[Symbol | String, untyped], ?estimated_total_tests_count: Integer, ?distributed: bool, ?local_test_suites_mode: bool) -> Datadog::CI::TestSession
+        def start_test_session: (?service: String?, ?tags: Hash[Symbol | String, untyped], ?estimated_total_tests_count: Integer, ?distributed: bool?, ?local_test_suites_mode: bool) -> Datadog::CI::TestSession
 
         def start_test_module: (String test_module_name, ?service: String?, ?tags: Hash[Symbol | String, untyped]) -> Datadog::CI::TestModule
 

sig/datadog/ci/test_tracing/null_component.rbs

@@ -11,7 +11,7 @@ module Datadog
         def trace: (String span_name, ?type: String, ?tags: Hash[untyped, untyped]) ?{ (Datadog::CI::Span span) -> untyped } -> untyped
 
         def start_test_session: (
-          ?service: String?, ?tags: Hash[untyped, untyped], ?estimated_total_tests_count: Integer, ?distributed: bool, ?local_test_suites_mode: bool
+          ?service: String?, ?tags: Hash[untyped, untyped], ?estimated_total_tests_count: Integer, ?distributed: bool?, ?local_test_suites_mode: bool
         ) -> nil
 
         def start_test_module: (String test_module_name, ?service: String?, ?tags: Hash[untyped, untyped]) -> nil

sig/datadog/ci/utils/file_storage.rbs

@@ -2,17 +2,27 @@ module Datadog
   module CI
     module Utils
       module FileStorage
+        class MissingNamespaceError < StandardError
+        end
+
         TEMP_DIR: String
+        ENV_NAMESPACE: String
 
         def self.store: (String key, Hash[Symbol, untyped] value) -> bool
 
         def self.retrieve: (String key) -> Hash[Symbol, untyped]?
 
-        def self.cleanup: () -> bool
+        def self.with_new_namespace: [T] () { (String) -> T } -> T
 
         def self.ensure_temp_dir_exists: () -> void
 
         def self.file_path_for: (String key) -> String
+
+        def self.storage_dir: (?String namespace) -> String
+
+        private
+
+        def self.cleanup: (String namespace) -> bool
       end
     end
   end