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



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

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



61
62
63
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/envelope_validator.rb', line 61

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<ElasticGraph::Indexer::Event>, Array<ElasticGraph::Indexer::MalformedEventError>)

Validates the envelope of each decoded JSON event and builds an Indexer::Event for each valid one.

Parameters:

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

    decoded JSON indexing events

Returns:

  • (Array(Array<ElasticGraph::Indexer::Event>, Array<ElasticGraph::Indexer::MalformedEventError>)) —

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



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

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:



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

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