Add a support for standard SSL configurations, proxy and basic auth configs.

[mashhurs]

Description

• Add SSL/TLS support for HTTPS schema registry connections
  • Add ssl_enabled option to enable/disable SSL
  • Add ssl_certificate and ssl_key options for PEM-based client authentication (unencrypted keys only)
  • Add ssl_certificate_authorities option for PEM-based server certificate validation
  • Add ssl_verification_mode option to control SSL verification (full, none)
  • Add ssl_cipher_suites option to configure cipher suites
  • Add ssl_supported_protocols option to configure TLS protocol versions (TLSv1.1, TLSv1.2, TLSv1.3)
  • Add ssl_truststore_path and ssl_truststore_password options for server certificate validation (JKS/PKCS12)
  • Add ssl_keystore_path and ssl_keystore_password options for mutual TLS authentication (JKS/PKCS12)
  • Add ssl_truststore_type and ssl_keystore_type options (JKS or PKCS12)
• Add HTTP proxy support with proxy option
• Add HTTP basic authentication support with username and password options

Integration test environment is same as https://github.com/logstash-plugins/logstash-integration-kafka

[kaisecheng]

I haven't finished the test yet. The main feedbacks are to add retry for calling schema registry and clarify if the plugin need to test against Kafka v3 and v4

[kaisecheng]

Added some suggestion for docs

[kaisecheng]

Regarding the test, can we add some tests for ssl_keystore_*?

[mashhurs]

Regarding the test, can we add some tests for ssl_keystore_*?

Added! This required to setup a mutual config in schema registry as well. Introduced new schema-registry-mutual.properties config with confluent.http.server.ssl.client.authentication=REQUIRED option.

[kaisecheng]

The second look mainly about the schema request handling and keeping the configs align with other plugins. Once they are addressed, I can approve it.

We need Karen to review the docs.

Also, if you decide to test against Kafka v4 only, please remove the pairs and add a comment "v3 is not supported in test"

[kaisecheng]

I am a bit confused now why non 200 need to raise StandardError. We want to retry 5xx and no retry for 4xx.
Here, all 4xx and 5xx are caught and no retry, right?

Also, I would like to reorganize the begin rescue block to improve readability.

def fetch_remote_schema(uri_string)
  client = nil
  client_options = {}
  ...
  max_retries = 3
  retry_count = 0
  
  begin
    response = client.get(uri_string).call
    
    unless response.code == 200
      @logger.error("Failed to fetch schema from #{uri_string}: #{response.code} - #{response.message}")
      raise BadResponseCodeError.new(response.code, response.message, uri_string)
    end
    
    body = response.body
    
    # Parse and extract schema
    parsed = JSON.parse(body)
    parsed.is_a?(Hash) && parsed.has_key?('schema') ? parsed['schema'] : body
  rescue JSON::ParserError
    return body
  rescue Manticore::ManticoreException, BadResponseCodeError => e
    # 4xx don't retry
    if e.is_a?(BadResponseCodeError) && e.code >= 400 && e.code < 500
      @logger.error()
      raise
    end

    # retry block
    retry_count += 1
    if retry_count <= max_retries
      ...
    end
  end
ensure
  client.close if client
end

[mashhurs]

Okay, so my confusion was if we introduce BadResponseCodeError this will anyway inherit StandardError and es-output has same class (similar approach we are following here) which inherits from LogStash::Error. In couple of places where LS interfaces with remote hosts like ES, we use LogStash::Error. From this point of view, it makes sense to introduce BadResponseCodeError and apply your point.

I have added with this commit - 86c5421

[kaisecheng]

@mashhurs The only unresolved is the retry logic. Others are good

[kaisecheng]

LGTM 🚀