Class: ElasticGraph::JSONIngestion::Indexer

Inherits:
Object
  • Object
show all
Defined in:
elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb

Overview

Adds JSON payload decoding to a format-neutral Indexer.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(indexer) ⇒ Indexer

Returns a new instance of Indexer.

Parameters:

  • indexer (ElasticGraph::Indexer) —

    the format-neutral indexer to wrap



40
41
42
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 40

def initialize(indexer)
  @indexer = indexer
end

Instance Attribute Details

#indexer ⇒ ElasticGraph::Indexer (readonly)

Returns the wrapped format-neutral indexer.

Returns:

  • (ElasticGraph::Indexer) —

    the wrapped format-neutral indexer



22
23
24
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 22

def indexer
  @indexer
end

Class Method Details

.from_parsed_yaml(parsed_yaml) {|Datastore::Client| ... } ⇒ Indexer

Builds a JSON-aware indexer from parsed YAML configuration.

Parameters:

  • parsed_yaml (Hash) —

    parsed YAML configuration

Yields:

  • (Datastore::Client) —

    optional block to customize the datastore client

Returns:



35
36
37
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 35

def self.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block)
  new(ElasticGraph::Indexer.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block))
end

Instance Method Details

#decode(payload) ⇒ Array<Hash<String, Object>>

Decodes one JSON Lines payload without processing it.

Parameters:

  • payload (String) —

    newline-delimited JSON indexing events

Returns:

  • (Array<Hash<String, Object>>) —

    decoded indexing events



77
78
79
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 77

def decode(payload)
  payload.split("\n").map { |event| ::JSON.parse(event) }
end

#logger ⇒ Logger

Returns:

  • (Logger) 


25
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 25

def logger = indexer.logger

#process(payload, refresh_indices: false) ⇒ void

This method returns an undefined value.

Decodes and processes one JSON Lines payload containing multiple indexing events.

If any events are invalid, an exception is raised, but valid events are still written to the datastore. No attempt is made to provide atomic "all or nothing" behavior.

Parameters:

  • payload (String) —

    newline-delimited JSON indexing events

  • refresh_indices (Boolean) (defaults to: false) —

    whether to synchronously refresh affected indices (intended for tests: this is dangerous to use in production)



52
53
54
55
56
57
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 52

def process(payload, refresh_indices: false)
  decoded_events = decode(payload)
  failures = process_returning_failures(decoded_events, refresh_indices: refresh_indices)
  return if failures.empty?
  raise ElasticGraph::Indexer::IndexingFailuresError.for(failures: failures, event_count: decoded_events.size)
end

#process_returning_failures(decoded_events, refresh_indices: false) ⇒ Array<ElasticGraph::Indexer::FailedEventError, ElasticGraph::Indexer::MalformedEventError>

Processes already-decoded events, returning individual failures. The caller is responsible for handling the failures.

This supports transports that must add metadata or combine several payloads before one bulk operation.

Parameters:

  • decoded_events (Array<Hash<String, Object>>) —

    decoded indexing events, as returned by #decode

  • refresh_indices (Boolean) (defaults to: false) —

    whether to synchronously refresh affected indices (intended for tests: this is dangerous to use in production)

Returns:

  • (Array<ElasticGraph::Indexer::FailedEventError, ElasticGraph::Indexer::MalformedEventError>) 


67
68
69
70
71
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 67

def process_returning_failures(decoded_events, refresh_indices: false)
  adapter = indexer.ingestion_adapters_by_format.fetch("json") # : ::ElasticGraph::JSONIngestion::IngestionAdapter
  events, malformed_failures = adapter.envelope_validator.events_from(decoded_events)
  processor.process_returning_failures(events, refresh_indices: refresh_indices) + malformed_failures
end

#processor ⇒ ElasticGraph::Indexer::Processor

Returns:

  • (ElasticGraph::Indexer::Processor) 


28
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer.rb', line 28

def processor = indexer.processor