<commit_before>require 'elasticsearch'
require 'aws-sdk-sqs'

module Workers
  class ElasticCloud
    attr_reader :input_queue_url, :output_cluster_url, :index_prefix, :document_type, :logger

    def initialize(input_queue_url, output_cluster_url,
                   index_prefix: 'ingest-measurements',
                   document_type: 'measurement',
                   logger: ::Logger.new($stdout))
      @input_queue_url = input_queue_url
      @output_cluster_url = output_cluster_url
      @index_prefix = index_prefix
      @document_type = document_type
      @logger = logger
      @logger.progname = self.class.to_s
    end

    def run
      logger.info("Starting real-time worker from #{input_queue_url} to #{output_cluster_uri.host}")
      setup_cluster

      poller = Aws::SQS::QueuePoller.new(input_queue_url)
      poller.poll do |message|
        body = JSON.parse(message.body)
        logger.info("Got message timestamped #{body['Timestamp']}")

        parsed_message = JSON.parse(body['Message'])
        logger.debug("Message version was #{parsed_message['version']}")

        payload = parsed_message['payload']

        now = Time.now.utc
        payload['@timestamp'] = now.to_i
        index_suffix = now.strftime('%Y-%m-%d')

        begin
          result = client.index(
              index: "#{index_prefix}-#{index_suffix}",
              type: document_type,
              body: payload
          )

          logger.info("Wrote #{result}")
        rescue Elasticsearch::Transport::Transport::Error => e
          logger.error("Unable to save measurement: #{e.message}")
          raise e
        end
      end
    end

    def setup_cluster
      client.indices.put_template(
          name: index_prefix,
          body: {
              template: index_prefix + '-*',
              order: 0,
              settings: {
                  number_of_shards: 1,
                  number_of_replicas: 0
              },
              mappings: {
                  _default_: {
                      _all: {
                          norms: false
                      },
                      _meta: {
                          version: '1.0.0'
                      },
                      date_detection: false,
                      dynamic_templates: [
                          {
                              strings_as_keyword: {
                                  mapping: {
                                      ignore_above: 1024,
                                      type: 'keyword'
                                  },
                                  match_mapping_type: 'string'
                              }
                          }
                      ],
                      properties: {
                          :@timestamp => {
                              type: 'date'
                          },
                          captured_at: {
                              type: 'date'
                          }
                      }
                  }
              }
          }
      )
    end

    def output_cluster_uri
      URI.parse(output_cluster_url)
    end

    def client<commit_msg>Fix up field mappings for correct date and location handling<commit_after>require 'elasticsearch'
require 'aws-sdk-sqs'

module Workers
  class ElasticCloud
    attr_reader :input_queue_url, :output_cluster_url, :index_prefix, :document_type, :logger

    def initialize(input_queue_url, output_cluster_url,
                   index_prefix: 'ingest-measurements',
                   document_type: 'measurement',
                   logger: ::Logger.new($stdout))
      @input_queue_url = input_queue_url
      @output_cluster_url = output_cluster_url
      @index_prefix = index_prefix
      @document_type = document_type
      @logger = logger
      @logger.progname = self.class.to_s
    end

    def run
      logger.info("Starting real-time worker from #{input_queue_url} to #{output_cluster_uri.host}")
      setup_cluster

      poller = Aws::SQS::QueuePoller.new(input_queue_url)
      poller.poll do |message|
        body = JSON.parse(message.body)
        logger.info("Got message timestamped #{body['Timestamp']}")

        parsed_message = JSON.parse(body['Message'])
        logger.debug("Message version was #{parsed_message['version']}")

        payload = parsed_message['payload']

        now = Time.now.utc
        payload['@timestamp'] = now.iso8601
        index_suffix = now.strftime('%Y-%m-%d')

        payload['ingest'] = {
            location: {
                lat: payload['loc_lat'],
                lon: payload['loc_lon'],
            }
        }

        begin
          result = client.index(
              index: "#{index_prefix}-#{index_suffix}",
              type: document_type,
              body: payload
          )

          logger.info("Wrote #{result}")
        rescue Elasticsearch::Transport::Transport::Error => e
          logger.error("Unable to save measurement: #{e.message}")
          raise e
        end
      end
    end

    def setup_cluster
      client.indices.put_template(
          name: index_prefix,
          body: {
              template: index_prefix + '-*',
              order: 0,
              settings: {
                  number_of_shards: 1,
                  number_of_replicas: 0
              },
              mappings: {
                  _default_: {
                      _all: {
                          norms: false
                      },
                      _meta: {
                          version: '1.0.0'
                      },
                      date_detection: false,
                      dynamic_templates: [
                          {
                              strings_as_keyword: {
                                  mapping: {
                                      ignore_above: 1024,
                                      type: 'keyword'
                                  },
                                  match_mapping_type: 'string'
                              }
                          }
                      ],
                      properties: {
                          :@timestamp => {
                              type: 'date'
                          },
                          service_uploaded: {
                              type: 'date'
                          },
                          :'ingest.location' => {
                              type: 'geo_point'
                          }
                      }
                  }
              }
          }
      )
    end

    def output_cluster_uri
      URI.parse(output_cluster_url)
    end

    def client
{'fraction_non_alphanumeric': 0.08726945569050833, 'fraction_numerical': 0.003598740440845704, 'mean_word_length': 1.1127652834969908, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2857055', 'n_tokens_mistral': 1697, 'n_tokens_neox': 1610, 'n_words': 303}