Add latest changes from gitlab-org/gitlab@master

3 files changed, 283 insertions, 0 deletions

diff --git a/app/services/pod_logs/base_service.rb b/app/services/pod_logs/base_service.rb
new file mode 100644
index 00000000000..668ee6b88a8
--- /dev/null
+++ b/app/services/pod_logs/base_service.rb
@@ -0,0 +1,132 @@
+# frozen_string_literal: true
+
+module PodLogs
+  class BaseService < ::BaseService
+    include ReactiveCaching
+    include Stepable
+
+    attr_reader :cluster, :namespace, :params
+
+    CACHE_KEY_GET_POD_LOG = 'get_pod_log'
+    K8S_NAME_MAX_LENGTH = 253
+
+    SUCCESS_RETURN_KEYS = %i(status logs pod_name container_name pods).freeze
+
+    def id
+      cluster.id
+    end
+
+    def initialize(cluster, namespace, params: {})
+      @cluster = cluster
+      @namespace = namespace
+      @params = filter_params(params.dup.stringify_keys).to_hash
+    end
+
+    def execute
+      with_reactive_cache(
+        CACHE_KEY_GET_POD_LOG,
+        namespace,
+        params
+      ) do |result|
+        result
+      end
+    end
+
+    def calculate_reactive_cache(request, _namespace, _params)
+      case request
+      when CACHE_KEY_GET_POD_LOG
+        execute_steps
+      else
+        exception = StandardError.new('Unknown reactive cache request')
+        Gitlab::ErrorTracking.track_and_raise_for_dev_exception(exception, request: request)
+        error(_('Unknown cache key'))
+      end
+    end
+
+    private
+
+    def valid_params
+      %w(pod_name container_name)
+    end
+
+    def check_arguments(result)
+      return error(_('Cluster does not exist')) if cluster.nil?
+      return error(_('Namespace is empty')) if namespace.blank?
+
+      success(result)
+    end
+
+    def check_param_lengths(_result)
+      pod_name = params['pod_name'].presence
+      container_name = params['container_name'].presence
+
+      if pod_name&.length.to_i > K8S_NAME_MAX_LENGTH
+        return error(_('pod_name cannot be larger than %{max_length}'\
+          ' chars' % { max_length: K8S_NAME_MAX_LENGTH }))
+      elsif container_name&.length.to_i > K8S_NAME_MAX_LENGTH
+        return error(_('container_name cannot be larger than'\
+          ' %{max_length} chars' % { max_length: K8S_NAME_MAX_LENGTH }))
+      end
+
+      success(pod_name: pod_name, container_name: container_name)
+    end
+
+    def get_raw_pods(result)
+      result[:raw_pods] = cluster.kubeclient.get_pods(namespace: namespace)
+
+      success(result)
+    end
+
+    def get_pod_names(result)
+      result[:pods] = result[:raw_pods].map(&:metadata).map(&:name)
+
+      success(result)
+    end
+
+    def check_pod_name(result)
+      # If pod_name is not received as parameter, get the pod logs of the first
+      # pod of this namespace.
+      result[:pod_name] ||= result[:pods].first
+
+      unless result[:pod_name]
+        return error(_('No pods available'))
+      end
+
+      unless result[:pods].include?(result[:pod_name])
+        return error(_('Pod does not exist'))
+      end
+
+      success(result)
+    end
+
+    def check_container_name(result)
+      pod_details = result[:raw_pods].first { |p| p.metadata.name == result[:pod_name] }
+      containers = pod_details.spec.containers.map(&:name)
+
+      # select first container if not specified
+      result[:container_name] ||= containers.first
+
+      unless result[:container_name]
+        return error(_('No containers available'))
+      end
+
+      unless containers.include?(result[:container_name])
+        return error(_('Container does not exist'))
+      end
+
+      success(result)
+    end
+
+    def pod_logs(result)
+      raise NotImplementedError
+    end
+
+    def filter_return_keys(result)
+      result.slice(*SUCCESS_RETURN_KEYS)
+    end
+
+    def filter_params(params)
+      params.slice(*valid_params)
+    end
+  end
+end
diff --git a/app/services/pod_logs/elasticsearch_service.rb b/app/services/pod_logs/elasticsearch_service.rb
new file mode 100644
index 00000000000..7524bf7ce10
--- /dev/null
+++ b/app/services/pod_logs/elasticsearch_service.rb
@@ -0,0 +1,63 @@
+# frozen_string_literal: true
+
+module PodLogs
+  class ElasticsearchService < BaseService
+    steps :check_arguments,
+          :check_param_lengths,
+          :get_raw_pods,
+          :get_pod_names,
+          :check_pod_name,
+          :check_container_name,
+          :check_times,
+          :check_search,
+          :pod_logs,
+          :filter_return_keys
+
+    self.reactive_cache_worker_finder = ->(id, _cache_key, namespace, params) { new(::Clusters::Cluster.find(id), namespace, params: params) }
+
+    private
+
+    def valid_params
+      %w(pod_name container_name search start end)
+    end
+
+    def check_times(result)
+      result[:start] = params['start'] if params.key?('start') && Time.iso8601(params['start'])
+      result[:end] = params['end'] if params.key?('end') && Time.iso8601(params['end'])
+
+      success(result)
+    rescue ArgumentError
+      error(_('Invalid start or end time format'))
+    end
+
+    def check_search(result)
+      result[:search] = params['search'] if params.key?('search')
+
+      success(result)
+    end
+
+    def pod_logs(result)
+      client = cluster&.application_elastic_stack&.elasticsearch_client
+      return error(_('Unable to connect to Elasticsearch')) unless client
+
+      result[:logs] = ::Gitlab::Elasticsearch::Logs.new(client).pod_logs(
+        namespace,
+        result[:pod_name],
+        result[:container_name],
+        result[:search],
+        result[:start],
+        result[:end]
+      )
+
+      success(result)
+    rescue Elasticsearch::Transport::Transport::ServerError => e
+      ::Gitlab::ErrorTracking.track_exception(e)
+
+      error(_('Elasticsearch returned status code: %{status_code}') % {
+        # ServerError is the parent class of exceptions named after HTTP status codes, eg: "Elasticsearch::Transport::Transport::Errors::NotFound"
+        # there is no method on the exception other than the class name to determine the type of error encountered.
+        status_code: e.class.name.split('::').last
+      })
+    end
+  end
+end
diff --git a/app/services/pod_logs/kubernetes_service.rb b/app/services/pod_logs/kubernetes_service.rb
new file mode 100644
index 00000000000..8f12b364e73
--- /dev/null
+++ b/app/services/pod_logs/kubernetes_service.rb
@@ -0,0 +1,88 @@
+# frozen_string_literal: true
+
+module PodLogs
+  class KubernetesService < BaseService
+    LOGS_LIMIT = 500.freeze
+    REPLACEMENT_CHAR = "\u{FFFD}"
+
+    EncodingHelperError = Class.new(StandardError)
+
+    steps :check_arguments,
+          :check_param_lengths,
+          :get_raw_pods,
+          :get_pod_names,
+          :check_pod_name,
+          :check_container_name,
+          :pod_logs,
+          :encode_logs_to_utf8,
+          :split_logs,
+          :filter_return_keys
+
+    self.reactive_cache_worker_finder = ->(id, _cache_key, namespace, params) { new(::Clusters::Cluster.find(id), namespace, params: params) }
+
+    private
+
+    def pod_logs(result)
+      result[:logs] = cluster.kubeclient.get_pod_log(
+        result[:pod_name],
+        namespace,
+        container: result[:container_name],
+        tail_lines: LOGS_LIMIT,
+        timestamps: true
+      ).body
+
+      success(result)
+    rescue Kubeclient::ResourceNotFoundError
+      error(_('Pod not found'))
+    rescue Kubeclient::HttpError => e
+      ::Gitlab::ErrorTracking.track_exception(e)
+
+      error(_('Kubernetes API returned status code: %{error_code}') % {
+        error_code: e.error_code
+      })
+    end
+
+    # Check https://gitlab.com/gitlab-org/gitlab/issues/34965#note_292261879
+    # for more details on why this is necessary.
+    def encode_logs_to_utf8(result)
+      return success(result) if result[:logs].nil?
+      return success(result) if result[:logs].encoding == Encoding::UTF_8
+
+      result[:logs] = encode_utf8(result[:logs])
+
+      success(result)
+    rescue EncodingHelperError
+      error(_('Unable to convert Kubernetes logs encoding to UTF-8'))
+    end
+
+    def split_logs(result)
+      result[:logs] = result[:logs].strip.lines(chomp: true).map do |line|
+        # message contains a RFC3339Nano timestamp, then a space, then the log line.
+        # resolution of the nanoseconds can vary, so we split on the first space
+        values = line.split(' ', 2)
+        {
+          timestamp: values[0],
+          message: values[1]
+        }
+      end
+
+      success(result)
+    end
+
+    def encode_utf8(logs)
+      utf8_logs = Gitlab::EncodingHelper.encode_utf8(logs.dup, replace: REPLACEMENT_CHAR)
+
+      # Gitlab::EncodingHelper.encode_utf8 can return '' or nil if an exception
+      # is raised while encoding. We prefer to return an error rather than wrongly
+      # display blank logs.
+      no_utf8_logs = logs.present? && utf8_logs.blank?
+      unexpected_encoding = utf8_logs&.encoding != Encoding::UTF_8
+
+      if no_utf8_logs || unexpected_encoding
+        raise EncodingHelperError, 'Could not convert Kubernetes logs to UTF-8'
+      end
+
+      utf8_logs
+    end
+  end
+end