Class: Logstash::Inputs::DynamoDB::DynamoDBLogParser

Inherits:
Object
  • Object
show all
Defined in:
lib/logstash/inputs/DynamoDBLogParser.rb

Constant Summary collapse

MAX_NUMBER_OF_BYTES_FOR_NUMBER = 
21

Instance Method Summary collapse

Constructor Details

#initialize(view_type, log_format, key_schema, region) ⇒ DynamoDBLogParser

Returns a new instance of DynamoDBLogParser.



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/logstash/inputs/DynamoDBLogParser.rb', line 35

def initialize(view_type, log_format, key_schema, region)
  @view_type = view_type
  @log_format = log_format
  @mapper ||= ObjectMapper.new()
  @mapper.setSerializationInclusion(JsonInclude::Include::NON_NULL)
  @mapper.addMixInAnnotations(AttributeValue, AttributeValueMixIn);
  @key_schema = key_schema
  ActiveSupport.encode_big_decimal_as_string = false
  @hash_template = Hash.new
  @hash_template["eventID"] = "0"
  @hash_template["eventName"] = "INSERT"
  @hash_template["eventVersion"] = "1.0"
  @hash_template["eventSource"] = "aws:dynamodb"
  @hash_template["awsRegion"] = region
end

Instance Method Details

#parse_scan(log, new_image_size) ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/logstash/inputs/DynamoDBLogParser.rb', line 52

def parse_scan(log, new_image_size)
  data_hash = JSON.parse(@mapper.writeValueAsString(log))

  @hash_template["dynamodb"] = Hash.new
  @hash_template["dynamodb"]["keys"] = Hash.new
  size_bytes = calculate_key_size_in_bytes(log)
  @key_schema.each { |x|
    @hash_template["dynamodb"]["keys"][x] = data_hash[x]
  }
  unless @view_type == "keys_only"
    size_bytes += new_image_size
    @hash_template["dynamodb"]["newImage"] = data_hash
  end
  @hash_template["dynamodb"]["sequenceNumber"] = "0"
  @hash_template["dynamodb"]["sizeBytes"] = size_bytes
  @hash_template["dynamodb"]["streamViewType"] = @view_type.upcase

  return parse_view_type(@hash_template)
end

#parse_stream(log) ⇒ Object 



73
74
75
 
# File 'lib/logstash/inputs/DynamoDBLogParser.rb', line 73

def parse_stream(log)
  return parse_view_type(JSON.parse(@mapper.writeValueAsString(log))["internalObject"])
end