Class: ElasticGraph::JSONIngestion::EnvelopeValidator

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

Overview

Turns decoded JSON payloads into indexing events, rejecting any whose envelope is malformed. Also owns the versioned JSON schema validators, which IngestionAdapter uses for record validation.

Instance Method Summary collapse

Constructor Details

#initialize(schema_artifacts:, logger:, configure_record_validator: nil) ⇒ EnvelopeValidator

Returns a new instance of EnvelopeValidator.

Parameters:

  • schema_artifacts (SchemaArtifacts::FromDisk) —

    the schema artifacts

  • logger (Logger) —

    the ElasticGraph logger

  • configure_record_validator (Proc, nil) (defaults to: nil) —

    optional callback to further configure the record validator



23
24
25
26
27
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/envelope_validator.rb', line 23

def initialize(schema_artifacts:, logger:, configure_record_validator: nil)
  @schema_artifacts = schema_artifacts
  @logger = logger
  @configure_record_validator = configure_record_validator
end

Instance Method Details

#closest_available_json_schema_version(requested_json_schema_version) ⇒ Integer?

The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match, the event can still be indexed.

This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version should be selected. So to get that behavior, the list is sorted in descending order.

Parameters:

  • requested_json_schema_version (Integer) —

    the version the event asked for

Returns:

  • (Integer, nil) —

    the closest available version, or nil when none is available



59
60
61
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/envelope_validator.rb', line 59

def closest_available_json_schema_version(requested_json_schema_version)
  @schema_artifacts.available_json_schema_versions.sort.reverse.min_by { |version| (requested_json_schema_version - version).abs }
end

#events_from(decoded_events) ⇒ Array(Array<Hash<String, Object>>, Array<ElasticGraph::Indexer::MalformedEventError>)

Validates the envelope of each decoded JSON event.

Parameters:

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

    decoded JSON indexing events

Returns:

  • (Array(Array<Hash<String, Object>>, Array<ElasticGraph::Indexer::MalformedEventError>)) —

    the events with a valid envelope, and a failure for each event with a malformed envelope



34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/envelope_validator.rb', line 34

def events_from(decoded_events)
  events, failures = [], []

  decoded_events.each do |decoded_event|
    if (failure = envelope_failure_for(decoded_event))
      failures << failure
    else
      events << event_from(decoded_event)
    end
  end

  [events, failures]
end

#validator(type, selected_json_schema_version) ⇒ ElasticGraph::Support::JSONSchema::Validator

Parameters:

  • type (String) —

    name of the type (or the event envelope) to validate

  • selected_json_schema_version (Integer) —

    the JSON schema version to validate against

Returns:



66
67
68
69
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/envelope_validator.rb', line 66

def validator(type, selected_json_schema_version)
  factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory
  factory.validator_for(type)
end