use PortManagementSupport mixin to reserve port in #register

[yaauie]

Uses the PortManagementSupport mixin from logstash-plugins/logstash-mixin-port_management_support#1 to reserve a port in #register so that the plugin will fail to start if the port is unavailable.

Resolves the TCP input's expression of elastic/logstash#17703

[jsvd]

I don't think we need to use blocks here since we're not wrapping behavior:

Suggested change

	@port_reservation = port_management.reserve(addr: @host, port: @port) do |reserved_addr, reserved_port|
	@loop = InputLoop.new(@id, reserved_addr, reserved_port, DecoderImpl.new(@codec, self), @tcp_keep_alive, java_ssl_context)
	end
	port_management.reserve(port: @port) # if this succeeds, there is a reservation for the port
	@loop = InputLoop.new(@id, @host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive, java_ssl_context)

Also we should set the reservation scope for the port alone. Not sure if it's worth differentiating the addr, we can be conservative here and allow the port to be reserved regardless of the addr.

[yaauie]

we're not wrapping behavior

We are.

PortManagementSupport::Reservation#initialize releases the reservation if an exception is raised by the block.

[yaauie]

Also we should set the reservation scope for the port alone. Not sure if it's worth differentiating the addr, we can be conservative here and allow the port to be reserved regardless of the addr.

In order to spawn the server that effectively holds the reservation, we need to know the addr to bind to, so really we are reserving an addr:port pair, not just the port.

[jsvd]

Could we just do:

Suggested change

	@port_reservation.convert do |reserved_addr, reserved_port|
	@logger.info("Starting tcp input listener", :address => "#{reserved_addr}:#{reserved_port}", :ssl_enabled => @ssl_enabled)
	@loop.start
	end
	@loop.wait_until_closed
	@logger.info("Starting tcp input listener", :address => "#{@host}:#{@port}", :ssl_enabled => @ssl_enabled)
	port_management.release(port: @port) { @loop.run }

Essentially tell the global manager:

1. ok I'm here.
2. release the port but don't free the reservation until I'm out?

[yaauie]

In Reservation#convert, the global mutex for creating reservations is held while the block is executed, and we don't want to hold that global lock for longer than necessary because other plugins using this mixin will be unable to create or convert while we hold the global lock.

The point of allowing the caller to execute a block while the global lock is held is to ensure that some other plugin can't use this library to create another reservation in the window between the dummy server being shut down and the caller standing up something to replace it.

If we were to add a second layer of locking (e.g., each reservation having its own mutex), we could do the bind outside of the global lock but we would introduce complexity around the race conditions.