Module: Smash::CloudPowers::Synapse::Pipe

Includes:
AwsResources, Helper, Zenv
Included in:
Smash::CloudPowers::SelfAwareness, Smash::CloudPowers::Synapse
Defined in:
lib/cloud_powers/synapse/pipe/pipe.rb

Instance Method Summary collapse

Methods included from Zenv

#env_vars, #file_tree_search, #i_vars, #project_root, #project_root=, #system_vars, #zfind

Methods included from Helper

#attr_map!, #available_resources, #called_from, #create_logger, #deep_modify_keys_with, #format_error_message, #log_file, #logger, #modify_keys_with, #smart_retry, #task_path, #task_require_path, #to_camel, #to_hyph, #to_i_var, #to_pascal, #to_ruby_file_name, #to_snake, #update_message_body, #valid_json?, #valid_url?

Methods included from AwsResources

#ec2, #image, #kinesis, #region, #s3, #sns, #sqs

Methods included from Auth

creds, region

Instance Method Details

#create_stream(name) ⇒ Object

Create a Kinesis stream or wait until the stream with the given name is through being created.

Parameters

  • name String

Returns Boolean or nil

  • returns true or false if the request was successful or not
  • returns true if the stream has already been created
  • returns false if the stream was not created


19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 19

def create_stream(name)
  begin
    config = stream_config(stream_name: name)
    resp = kinesis.create_stream(config)
    kinesis.wait_until(:stream_exists, stream_name: config[:stream_name])
    resp.successful? # (http request successful && stream created)?
  rescue Exception => e
    if e.kind_of? Aws::Kinesis::Errors::ResourceInUseException
      logger.info "#{name} already created"
      return if stream_status(name) == 'ACTIVE'
      logger.info "Not ready for traffic.  Wait for 30 seconds..."
      sleep 1
      true # acts like it would if it had to create the stream
    else
      error_message = format_error_message(e)
      logger.error error_message
      false # the request was not successful
    end
  end
end

#flow_from_pipe(stream) ⇒ Object

Use the KCL and LangDaemon to read from a stream

Parameters stream String

Notes This method is not implemented yet (V 0.2.7)



46
47
48
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 46

def flow_from_pipe(stream)
  throw NotImplementedError
end

#flow_to_pipe(stream) ⇒ Object

Sends data through a Pipe. This method is used for lower throughput applications, e.g. logging, status updates

Parameters

  • stream String

Returns Notes This method is not implemented yet (V 0.2.7)



61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 61

def flow_to_pipe(stream)
  throw NotImplementedError
  create_stream(stream) unless stream_exists? stream
  records = yield if block_given?
  body = message_body_collection(records)
  # TODO: this isn't working yet.  figure out retry logic
  # resp = kinesis.put_records(body)
  # retry(lambda { stream_exists? stream }) flow_to(stream)
  @last_sequence_number = resp.records.map(&:sequence_number).sort.last
  # TODO: what to return? true?
end

#from_pipe(stream) ⇒ Object

Read messages from the Pipe without using the KCL

Parameters stream String

Notes This method is not implemented yet (V 0.2.7)



79
80
81
82
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 79

def from_pipe(stream)
  # implement get_records and/or other consuming app stuff
  throw NotImplementedError
end

#message_body_collection(records) ⇒ Object

This message will prepare a set of collections to be sent through the Pipe

Parameters

  • records

Notes This method is not implemented yet (V 0.2.7)



91
92
93
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 91

def message_body_collection(records)
  throw NotImplementedError
end

#pipe_message_body(opts = {}) ⇒ Object

Default message package. This method yields the basic configuration and message body for a stream and all options can be changed.

Parameters opts Hash (optional)

  • stream_name: String name of the stream to pipe to
  • data: String message to send
  • partition_key: String defaults to @instance_id

Returns Hash

Notes:

  • See #zfind()
  • See #instance_id()
  • See #update_message_body()


110
111
112
113
114
115
116
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 110

def pipe_message_body(opts = {})
  {
    stream_name:      zfind(opts[:stream_name]) || zfind('status_stream'),
    data:             opts[:data] || update_message_body(opts),
    partition_key:    opts[:partition_key] || @instance_id || 'unk'
  }
end

#pipe_to(stream) ⇒ Object

Use Kinesis streams to send a message. The message is given to the method through a block that gets passed to the method.

Parameters

  • stream String
  • block - a block that generates a string that will be used in the message body

Returns the sequence_number from the sent message.

Example use pipe_to(:status_stream) do # the return from the inner method is what is sent do_some_stuff_to_generate_a_message() end



133
134
135
136
137
138
139
140
141
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 133

def pipe_to(stream)
  message = ''
  create_stream(stream) unless stream_exists? stream
  message = yield if block_given?
  body = update_message_body(message)
  resp = kinesis.put_record pipe_message_body(stream_name: stream, data: body.to_json)
  # TODO: implement retry logic for failed request
  @last_sequence_number = resp.sequence_number
end

#stream_config(opts = {}) ⇒ Object

New stream config with sensible defaults

Parameters

  • opts Hash (optional)
    • stream_name - the name to give the stream
    • shard_count - the number of shards to create


149
150
151
152
153
154
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 149

def stream_config(opts = {})
  config = {
    stream_name: opts[:stream_name] || zfind(:status_stream),
    shard_count: opts[:shard_count] || 1
  }
end

#stream_exists?(name) ⇒ Boolean

Find out if the stream already exists.

Parameters

  • name String

Returns Boolean

Returns:

  • (Boolean) 


163
164
165
166
167
168
169
170
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 163

def stream_exists?(name)
  begin
    kinesis.describe_stream(stream_name: name)
    true
  rescue Aws::Kinesis::Errors::ResourceNotFoundException => e
    false
  end
end

#stream_status(name) ⇒ Object

Get the status name for this stream

Parameters *name String

Returns stream status, one of: CREATING, DELETING, ACTIVE, UPDATING



179
180
181
 
# File 'lib/cloud_powers/synapse/pipe/pipe.rb', line 179

def stream_status(name)
  kinesis.describe_stream(stream_name: name).stream_description.stream_status
end