Pluginmanager clean after mutate (#17203)

* pluginmanager: always clean after mutate

* pluginmanager: don't skip updating plugins installed with --version

* pr feedback

(cherry picked from commit 8c96913)

Author: yaauie

lib/pluginmanager/command.rb

@@ -47,6 +47,31 @@ def remove_unused_locally_installed_gems!
     end
   end
 
+  def remove_orphan_dependencies!
+    locked_gem_names = ::Bundler::LockfileParser.new(File.read(LogStash::Environment::LOCKFILE)).specs.map(&:full_name).to_set
+    orphan_gem_specs = ::Gem::Specification.each
+                                           .reject(&:stubbed?) # skipped stubbed (uninstalled) gems
+                                           .reject(&:default_gem?) # don't touch jruby-included default gems
+                                           .reject{ |spec| locked_gem_names.include?(spec.full_name) }
+                                           .sort
+
+    inactive_plugins, orphaned_dependencies = orphan_gem_specs.partition { |spec| LogStash::PluginManager.logstash_plugin_gem_spec?(spec) }
+
+    # uninstall plugins first, to limit damage should one fail to uninstall
+    inactive_plugins.each { |plugin| uninstall_gem!("inactive plugin", plugin) }
+    orphaned_dependencies.each { |dep| uninstall_gem!("orphaned dependency", dep) }
+  end
+
+  def uninstall_gem!(desc, spec)
+    require "rubygems/uninstaller"
+    Gem::DefaultUserInteraction.use_ui(debug? ? Gem::DefaultUserInteraction.ui : Gem::SilentUI.new) do
+      Gem::Uninstaller.new(spec.name, version: spec.version, force: true, executables: true).uninstall
+    end
+    puts "cleaned #{desc} #{spec.name} (#{spec.version})"
+  rescue Gem::InstallError => e
+    report_exception("Failed to uninstall `#{spec.full_name}`", e)
+  end
+
   def relative_path(path)
     require "pathname"
     ::Pathname.new(path).relative_path_from(::Pathname.new(LogStash::Environment::LOGSTASH_HOME)).to_s

lib/pluginmanager/gemfile.rb

@@ -133,8 +133,8 @@ def add_gem(_gem)
     # update existing or add new
     def update_gem(_gem)
       if old = find_gem(_gem.name)
-        # always overwrite requirements if specified
-        old.requirements = _gem.requirements unless no_constrains?(_gem.requirements)
+        # always overwrite requirements
+        old.requirements = _gem.requirements
         # but merge options
         old.options = old.options.merge(_gem.options)
       else

lib/pluginmanager/install.rb

@@ -79,6 +79,7 @@ def execute
     install_gems_list!(gems)
     remove_unused_locally_installed_gems!
     remove_unused_integration_overlaps!
+    remove_orphan_dependencies!
   end
 
   private

lib/pluginmanager/remove.rb

@@ -67,6 +67,7 @@ def execute
     exit(1) unless ::Bundler::LogstashUninstall.uninstall!(plugin_list)
     LogStash::Bundler.genericize_platform
     remove_unused_locally_installed_gems!
+    remove_orphan_dependencies!
   rescue => exception
     report_exception("Operation aborted, cannot remove plugin", exception)
   end

lib/pluginmanager/update.rb

@@ -95,10 +95,8 @@ def update_gems!
       output << LogStash::Bundler.genericize_platform unless output.nil?
     end
 
-    # We currently dont removed unused gems from the logstash installation
-    # see: https://github.com/elastic/logstash/issues/6339
-    # output = LogStash::Bundler.invoke!(:clean => true)
     display_updated_plugins(previous_gem_specs_map)
+    remove_orphan_dependencies!
   rescue => exception
     gemfile.restore!
     report_exception("Updated Aborted", exception)

qa/integration/fixtures/update_spec.yml

@@ -0,0 +1,3 @@
+---
+services:
+  - logstash

qa/integration/services/logstash_service.rb

@@ -306,7 +306,7 @@ def run_cmd(cmd_args, change_dir = true, environment = {})
     if ENV.key?("BUILD_JAVA_HOME") && !process.environment.key?("LS_JAVA_HOME")
       process.environment["LS_JAVA_HOME"] = ENV["BUILD_JAVA_HOME"]
     end
-    process.io.stdout = process.io.stderr = out
+    process.io.stdout = process.io.stderr = SynchronizedDelegate.new(out)
 
     Bundler.with_unbundled_env do
       if change_dir
@@ -327,6 +327,31 @@ def run(*args)
     run_cmd [@logstash_bin, *args]
   end
 
+  ##
+  # A `SynchronizedDelegate` wraps any object and ensures that exactly one
+  # calling thread is invoking methods on it at a time. This is useful for our
+  # clumsy setting of process io STDOUT and STDERR to the same IO object, which
+  # can cause interleaved writes.
+  class SynchronizedDelegate
+    def initialize(obj)
+      require "monitor"
+      @mon = Monitor.new
+      @obj = obj
+    end
+
+    def respond_to_missing?(method_name, include_private = false)
+      @obj.respond_to?(method_name, include_private) || super
+    end
+
+    def method_missing(method_name, *args, &block)
+      return super unless @obj.respond_to?(method_name)
+
+      @mon.synchronize do
+        @obj.public_send(method_name, *args, &block)
+      end
+    end
+  end
+
   class PluginCli
 
     LOGSTASH_PLUGIN = File.join("bin", "logstash-plugin")
@@ -360,9 +385,26 @@ def list(*plugins, verbose: false)
       run(command)
     end
 
-    def install(plugin_name, *additional_plugins)
-      plugin_list = ([plugin_name]+additional_plugins).flatten
-      run("install #{Shellwords.shelljoin(plugin_list)}")
+    def install(plugin_name, *additional_plugins, version: nil, verify: true, preserve: false, local: false)
+      args = []
+      args << "--no-verify" unless verify
+      args << "--preserve" if preserve
+      args << "--local" if local
+      args << "--version" << version unless version.nil?
+      args.concat(([plugin_name]+additional_plugins).flatten)
+
+      run("install #{Shellwords.shelljoin(args)}")
+    end
+
+    def update(*plugin_list, level: nil, local: nil, verify: nil, conservative: nil)
+      args = []
+      args << (verify ? "--verify" : "--no-verify") unless verify.nil?
+      args << "--level" << "#{level}" unless level.nil?
+      args << "--local" if local
+      args << (conservative ? "--conservative" : "--no-conservative") unless conservative.nil?
+      args.concat(plugin_list)
+
+      run("update #{Shellwords.shelljoin(args)}")
     end
 
     def run(command)

qa/integration/specs/cli/install_spec.rb

@@ -19,6 +19,7 @@
 require_relative "../../framework/settings"
 require_relative "../../services/logstash_service"
 require_relative "../../framework/helpers"
+require_relative "pluginmanager_spec_helper"
 require "logstash/devutils/rspec/spec_helper"
 require "stud/temporary"
 require "fileutils"
@@ -29,23 +30,32 @@ def gem_in_lock_file?(pattern, lock_file)
   content.match(pattern)
 end
 
+def plugin_filename_re(name, version)
+  %Q(\b#{Regexp.escape name}-#{Regexp.escape version}(-java)?\b)
+end
+
 # Bundler can mess up installation successful output: https://github.com/elastic/logstash/issues/15801
 INSTALL_SUCCESS_RE = /IB?nstall successful/
 INSTALLATION_SUCCESS_RE = /IB?nstallation successful/
 
+INSTALLATION_ABORTED_RE = /Installation aborted/
+
 describe "CLI > logstash-plugin install" do
-  before(:all) do
+  before(:each) do
     @fixture = Fixture.new(__FILE__)
     @logstash = @fixture.get_service("logstash")
     @logstash_plugin = @logstash.plugin_cli
-    @pack_directory =  File.expand_path(File.join(File.dirname(__FILE__), "..", "..", "fixtures", "logstash-dummy-pack"))
   end
 
   shared_examples "install from a pack" do
     let(:pack) { "file://#{File.join(@pack_directory, "logstash-dummy-pack.zip")}" }
     let(:install_command) { "bin/logstash-plugin install" }
     let(:change_dir) { true }
 
+    before(:all) do
+      @pack_directory =  File.expand_path(File.join(File.dirname(__FILE__), "..", "..", "fixtures", "logstash-dummy-pack"))
+    end
+
     # When you are on anything by linux we won't disable the internet with seccomp
     if RbConfig::CONFIG["host_os"] == "linux"
       context "without internet connection (linux seccomp wrapper)" do
@@ -152,4 +162,92 @@ def gem_in_lock_file?(pattern, lock_file)
       end
     end
   end
+
+  context "rubygems hosted plugin" do
+    include_context "pluginmanager validation helpers"
+    shared_examples("overwriting existing") do
+      before(:each) do
+        aggregate_failures("precheck") do
+          expect("#{plugin_name}-#{existing_plugin_version}").to_not be_installed_gem
+          expect("#{plugin_name}-#{specified_plugin_version}").to_not be_installed_gem
+        end
+        aggregate_failures("setup") do
+          execute = @logstash_plugin.install(plugin_name, version: existing_plugin_version)
+
+          expect(execute.stderr_and_stdout).to match(INSTALLATION_SUCCESS_RE)
+          expect(execute.exit_code).to eq(0)
+
+          expect("#{plugin_name}-#{existing_plugin_version}").to be_installed_gem
+          expect("#{plugin_name}-#{specified_plugin_version}").to_not be_installed_gem
+        end
+      end
+      it "installs the specified version and removes the pre-existing one" do
+        execute = @logstash_plugin.install(plugin_name, version: specified_plugin_version)
+
+        aggregate_failures("command execution") do
+          expect(execute.stderr_and_stdout).to match(INSTALLATION_SUCCESS_RE)
+          expect(execute.exit_code).to eq(0)
+        end
+
+        installed = @logstash_plugin.list(plugin_name, verbose: true)
+        expect(installed.stderr_and_stdout).to match(/#{Regexp.escape plugin_name} [(]#{Regexp.escape(specified_plugin_version)}[)]/)
+
+        expect("#{plugin_name}-#{existing_plugin_version}").to_not be_installed_gem
+        expect("#{plugin_name}-#{specified_plugin_version}").to be_installed_gem
+      end
+    end
+
+    context "when installing over an older version" do
+      let(:plugin_name) { "logstash-filter-qatest" }
+      let(:existing_plugin_version) { "0.1.0" }
+      let(:specified_plugin_version) { "0.1.1" }
+
+      include_examples "overwriting existing"
+    end
+
+    context "when installing over a newer version" do
+      let(:plugin_name) { "logstash-filter-qatest" }
+      let(:existing_plugin_version) { "0.1.0" }
+      let(:specified_plugin_version) { "0.1.1" }
+
+      include_examples "overwriting existing"
+    end
+
+    context "installing plugin that isn't present" do
+      it "installs the plugin" do
+        aggregate_failures("prevalidation") do
+          expect("logstash-filter-qatest").to_not be_installed_gem
+        end
+
+        execute = @logstash_plugin.install("logstash-filter-qatest")
+
+        expect(execute.stderr_and_stdout).to match(INSTALLATION_SUCCESS_RE)
+        expect(execute.exit_code).to eq(0)
+
+        installed = @logstash_plugin.list("logstash-filter-qatest")
+        expect(installed.stderr_and_stdout).to match(/logstash-filter-qatest/)
+        expect(installed.exit_code).to eq(0)
+
+        expect(gem_in_lock_file?(/logstash-filter-qatest/, @logstash.lock_file)).to be_truthy
+
+        expect("logstash-filter-qatest").to be_installed_gem
+      end
+    end
+    context "installing plugin that doesn't exist on rubygems" do
+      it "doesn't install anything" do
+        execute = @logstash_plugin.install("logstash-filter-404-no-exist")
+
+        expect(execute.stderr_and_stdout).to match(INSTALLATION_ABORTED_RE)
+        expect(execute.exit_code).to eq(1)
+      end
+    end
+    context "installing gem that isn't a plugin" do
+      it "doesn't install anything" do
+        execute = @logstash_plugin.install("dummy_gem")
+
+        expect(execute.stderr_and_stdout).to match(INSTALLATION_ABORTED_RE)
+        expect(execute.exit_code).to eq(1)
+      end
+    end
+  end
 end

qa/integration/specs/cli/pluginmanager_spec_helper.rb

@@ -0,0 +1,59 @@
+require 'pathname'
+
+shared_context "pluginmanager validation helpers" do
+
+  matcher :be_installed_gem do
+    match do |actual|
+      common(actual)
+      @gemspec_present && @gem_installed
+    end
+
+    match_when_negated do |actual|
+      common(actual)
+      !@gemspec_present && !@gem_installed
+    end
+
+    define_method :common do |actual|
+      version_suffix = /-[0-9.]+(-java)?$/
+      filename_matcher = actual.match?(version_suffix) ? actual : /^#{Regexp.escape(actual)}#{version_suffix}/
+
+      @gems = (logstash_gemdir / "gems").glob("*-*")
+      @gemspecs = (logstash_gemdir / "specifications").glob("*-*.gemspec")
+
+      @gem_installed = @gems.find { |gem| gem.basename.to_s.match?(filename_matcher) }
+      @gemspec_present = @gemspecs.find { |gemspec| gemspec.basename(".gemspec").to_s.match?(filename_matcher) }
+    end
+
+    failure_message do |actual|
+      reasons = []
+      reasons << "the gem dir could not be found (#{@gems})" unless @gem_installed
+      reasons << "the gemspec could not be found (#{@gemspecs})" unless @gemspec_present
+
+      "expected that #{actual} would be installed, but #{reasons.join(' and ')}"
+    end
+    failure_message_when_negated do |actual|
+      reasons = []
+      reasons << "the gem dir is present (#{@gem_installed})" if @gem_installed
+      reasons << "the gemspec is present (#{@gemspec_present})" if @gemspec_present
+
+      "expected that #{actual} would not be installed, but #{reasons.join(' and ')}"
+    end
+  end
+
+  def logstash_home
+    return super() if defined?(super)
+    return @logstash.logstash_home if @logstash
+    fail("no @logstash, so we can't get logstash_home")
+  end
+
+  def logstash_gemdir
+    pathname_base = (Pathname.new(logstash_home) / "vendor" / "bundle" / "jruby")
+    candidate_dirs = pathname_base.glob("[0-9]*")
+    case candidate_dirs.size
+    when 0 then fail("no version dir found in #{pathname_base}")
+    when 1 then candidate_dirs.first
+    else
+      fail("multiple version dirs found in #{pathname_base} (#{candidate_dirs.map(&:basename)}")
+    end
+  end
+end