Fix RangeError by converting large integers to BSON::Decimal128

lib/fluent/plugin/out_mongo.rb

@@ -351,6 +351,9 @@ def operate(database, collection, records)
             replace_key_of_hash(r, /^\$/, @replace_dollar_in_key_with)
           end
         end
+        records.map! do |r|
+          replace_value_of_hash(r)
+        end
 
         get_collection(database, collection, @collection_options).insert_many(records)
       rescue Mongo::Error::BulkWriteError => e
@@ -384,5 +387,23 @@ def replace_key_of_hash(hash_or_array, pattern, replacement)
         hash_or_array
       end
     end
+
+    INT64_MAX = (2 ** 63) - 1
+    def replace_value_of_hash(hash_or_array)
+      case hash_or_array
+      when Array
+        hash_or_array.map { |v| replace_value_of_hash(v) }
+      when Hash
+        hash_or_array.each_pair { |k, v| hash_or_array[k] = replace_value_of_hash(v) }
+      when Integer
+        if hash_or_array > INT64_MAX
+          BSON::Decimal128.new(hash_or_array.to_s)
+        else
+          hash_or_array
+        end
+      else
+        hash_or_array
+      end
+    end
   end
 end

test/plugin/test_out_mongo.rb

@@ -249,6 +249,20 @@ def test_write_with_expire_index
     assert_equal({"expireAfterSeconds"=>120.0}, expire_after_hash)
   end
 
+  def test_overflow_integer_value
+    d = create_driver
+    d.run(default_tag: 'test') do
+      time = event_time("2011-01-02 13:14:15 UTC")
+      d.feed(time, {'overflow' => (2 ** 63)})
+      d.feed(time, {'nested' => {'overflow' => (2 ** 63)}})
+      d.feed(time, {'array' => [(2 ** 63)]})
+    end
+    documents = get_documents
+    assert_equal(BSON::Decimal128.new((2 ** 63).to_s), documents[0]['overflow'])
+    assert_equal(BSON::Decimal128.new((2 ** 63).to_s), documents[1]['nested']['overflow'])
+    assert_equal(BSON::Decimal128.new((2 ** 63).to_s), documents[2]['array'][0])
+  end
+
   class WriteWithCollectionPlaceholder < self
     def setup
       @tag = 'custom'