[logstash-plugins] moved license mention in files; various bugfixes and hardening

integration/logstash-plugins/README.md

@@ -1,7 +1,5 @@
 # Logstash Plugins for Interacting with Vespa
 
-This repository contains plugins to both read from and write to Vespa by using Logstash.
+Plugins to both read from and write to Vespa by using Logstash.
 
-Check out the README.md files in the respective plugin directories for more information.
-
-This repository doesn't have issues enabled, but feel free to open related issues at https://github.com/vespa-engine/vespa
+Check out the README.md files in the respective plugin directories for more information.

integration/logstash-plugins/logstash-input-vespa/CHANGELOG.md

@@ -1,4 +1,6 @@
+## 0.2.0
+Added support for mTLS certificates, selector, page_size, backend_concurrency, timeout, from_timestamp, and to_timestamp
+
 ## 0.1.0
 
 Initial version. Can connect to an HTTP endpoint of Vespa and pull documents via Visit and use the continuation token
-

integration/logstash-plugins/logstash-input-vespa/Gemfile

@@ -9,3 +9,11 @@ if Dir.exist?(logstash_path) && use_logstash_source
   gem 'logstash-core', :path => "#{logstash_path}/logstash-core"
   gem 'logstash-core-plugin-api', :path => "#{logstash_path}/logstash-core-plugin-api"
 end
+
+group :development, :test do
+  gem 'logstash-devutils'
+  gem 'webmock'
+  gem 'rspec'
+end
+
+gem 'minitar', '~> 1.0'

integration/logstash-plugins/logstash-input-vespa/README.md

@@ -9,6 +9,17 @@ Download and unpack/install Logstash, then:
 bin/logstash-plugin install logstash-input-vespa
 ```
 
+## Development
+
+To run tests, you'll need to clone the Logstash branch you're developing the plugin for. See https://github.com/elastic/logstash
+
+Then:
+```
+export LOGSTASH_PATH=/path/to/logstash/repository/clone
+export LOGSTASH_SOURCE=1
+bundle exec rspec
+```
+
 ## Usage
 
 Minimal Logstash config example:

integration/logstash-plugins/logstash-input-vespa/lib/logstash/inputs/vespa.rb

@@ -1,4 +1,7 @@
 # encoding: utf-8
+
+# Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
 require "logstash/inputs/base"
 require "logstash/namespace"
 require "net/http"
@@ -22,6 +25,12 @@ class LogStash::Inputs::Vespa < LogStash::Inputs::Base
   # The cluster parameter to use in the request.
   config :cluster, :validate => :string, :required => true
 
+  # Maximum number of retries for failed HTTP requests
+  config :max_retries, :validate => :number, :default => 3
+
+  # Delay in seconds for the first retry attempt. We double this delay for each subsequent retry.
+  config :retry_delay, :validate => :number, :default => 1
+
   # Path to the client certificate file for mTLS.
   config :client_cert, :validate => :path
 
@@ -85,6 +94,9 @@ def run(queue)
     uri.query = URI.encode_www_form(@uri_params)
     continuation = nil
 
+    retries = 0
+    current_delay = @retry_delay
+
     loop do
       response = fetch_documents_from_vespa(uri)
       # response should look like:
@@ -120,28 +132,58 @@ def run(queue)
         end
 
       else
-        @logger.error("Failed to fetch documents from Vespa", :request => uri.to_s,
+        # Handle retriable status codes (5xx)
+        if (500..599).include?(response.code.to_i) && retries < (@max_retries - 1)
+          retries += 1
+          @logger.warn("Retriable error from Vespa, retrying", 
+                      :response_code => response.code,
+                      :retry_count => retries,
+                      :max_retries => @max_retries,
+                      :next_retry_delay => current_delay)
+          sleep(current_delay)
+          current_delay *= 2
+        else
+          @logger.error("Failed to fetch documents from Vespa", :request => uri.to_s,
                       :response_code => response.code, :response_message => response.message)
-        break # TODO retry? Only on certain codes?
+          break
+        end
       end # if response.is_a?(Net::HTTPSuccess)
 
     end # loop do
   end # def run
 
   def fetch_documents_from_vespa(uri)
-    http = Net::HTTP.new(uri.host, uri.port)
-    if uri.scheme == "https"
-      http.use_ssl = true
-      http.cert = @cert
-      http.key = @key
-      http.verify_mode = OpenSSL::SSL::VERIFY_PEER
+    retries = 0
+    current_delay = @retry_delay  # Start with the initial delay
+
+    begin
+      http = Net::HTTP.new(uri.host, uri.port)
+      if uri.scheme == "https"
+        http.use_ssl = true
+        http.cert = @cert
+        http.key = @key
+        http.verify_mode = OpenSSL::SSL::VERIFY_PEER
+      end
+
+      request = Net::HTTP::Get.new(uri.request_uri)
+      http.request(request)
+    rescue => e
+      retries += 1
+      if retries < @max_retries
+        @logger.warn("Failed to make HTTP request to Vespa, retrying", 
+                    :error => e.message, 
+                    :retry_count => retries, 
+                    :max_retries => @max_retries,
+                    :next_retry_delay => current_delay)
+        sleep(current_delay)
+        current_delay *= 2  # Double the delay for next retry
+        retry
+      else
+        @logger.error("Failed to make HTTP request to Vespa after #{@max_retries} attempts", 
+                     :error => e.message)
+        nil
+      end
     end
-
-    request = Net::HTTP::Get.new(uri.request_uri)
-    http.request(request)
-  rescue => e
-      @logger.error("Failed to make HTTP request to Vespa", :error => e.message)
-      nil
   end # def fetch_documents_from_vespa
 
   def parse_response(response)

integration/logstash-plugins/logstash-input-vespa/logstash-input-vespa.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-input-vespa'
-  s.version         = '0.2.0'
+  s.version         = '0.3.0'
   s.licenses = ['Apache-2.0']
   s.summary = "Logstash input plugin reading from Vespa"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
@@ -18,8 +18,13 @@ Gem::Specification.new do |s|
   s.metadata = { "logstash_plugin" => "true", "logstash_group" => "input" }
 
   # Gem dependencies
-  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
+  s.add_runtime_dependency "logstash-core", ">= 8.0.0"
   s.add_runtime_dependency 'logstash-codec-plain'
   s.add_runtime_dependency 'stud', '>= 0.0.22'
+  s.add_runtime_dependency 'logstash-codec-json'
+
+  # Development dependencies
   s.add_development_dependency 'logstash-devutils'
+  s.add_development_dependency 'rspec', '~> 3.0'
+  s.add_development_dependency 'webmock', '~> 3.0'
 end